[subscriber] fix replay dedup, instance identity, and error classification from review

Author: Martin7-1

subscriber/AGENTS.md

@@ -26,18 +26,19 @@ uv run ruff format --check subscriber/ tests/ # 仅检查，不修改
 # 类型检查（mypy strict）
 uv run mypy subscriber/
 
-# 启动进程（默认连接本机 vLLM）
-uv run python -m subscriber
+# 启动进程（SPECTRUM_DEPLOYMENT_NAME 与 --kvcm-base-url 为启动必填）
+export SPECTRUM_DEPLOYMENT_NAME=my-deployment
+uv run python -m subscriber --kvcm-base-url http://10.0.0.1:8080
 
 # 指定参数启动
-uv run python -m subscriber \
+SPECTRUM_DEPLOYMENT_NAME=my-deployment uv run python -m subscriber \
   --zmq-pub-endpoint tcp://localhost:5557 \
   --zmq-replay-endpoint tcp://localhost:5558 \
-  --kvcm-addr 10.0.0.1:50051 \
+  --kvcm-base-url spectrum://vs-example:6382 \
   --engine-type vllm
 
-# 使用配置文件
-uv run python -m subscriber --config config.yaml
+# 使用配置文件（kvcm_base_url 可写在 yaml 中）
+SPECTRUM_DEPLOYMENT_NAME=my-deployment uv run python -m subscriber --config config.yaml
 ```
 
 ## 质量门禁（Quality Gates）
@@ -87,8 +88,9 @@ proto/client 测试和 `uv run mypy subscriber/`，确认 protobuf 3.20.3
 - **发送门控**：向 kvcm 转发实时或 replay batch 前必须通过 `EngineHealthCoordinator.wait_ready_epoch()` 获取当前可发送 epoch。
 - **分阶段 span 计时**：从 event 到手到发送 kvcm 的每个阶段用 span 埋点，可拓展。载体 `EngineEventBatch(batches, timer)` 由 adapter yield：timer 起源于 adapter 开始处理该 event，adapter 标记自身阶段（live=`decode`、replay=`replay_fetch`），`send_incremental_events` 在同一 timer 上续标 `queue_wait`/`gate_wait`/`block_filter`/`metadata_learn`/`kvcm_send`。加新阶段只需在相应位置多一次 `timer.mark("stage")`。span 上报是 best-effort，`StageTimer.mark/spans` 与 `report()` 契约永不抛异常，绝不阻断转发主路径。
 - **Metrics 上报规范**：所有指标通过 `subscriber/metrics/` 包（`_base.py` 中的 `_dashlog_counter` / `_dashlog_gauge`）统一出口上报，自动添加 `kvcache_subscriber_` 前缀。维度信息（endpoint、status、reason 等）通过 `tags` 参数传递（对应 Prometheus labels），不要编码进 metric name。dashlog 本地不可编译时所有 report 函数为 no-op（`try: import dashlog / except ImportError` fallback）。新增或修改指标后必须同步更新 `docs/metrics.json` 指标目录。
-- **Replay 语义**：检测到 seq gap 时，先聚合所有 replay batch 一次性 yield，再 yield 当前实时消息。replay 使用 DEALER socket（非 REQ），避免 async 下严格一问一答的阻塞问题。
-- **配置优先级**：CLI 显式参数 > `--config yaml` 文件 > dataclass 默认值。
+- **Replay 语义**：检测到 seq gap 时，先聚合 replay batch 一次性 yield，再 yield 当前实时消息。vLLM replay 端会返回所有 `seq >= gap_start_seq` 的缓冲消息（包含触发 gap 的当前实时消息及更新消息），必须过滤只转发缺失区间 `[gap_start_seq, current_seq - 1]`（其余会以实时消息到达，直接转发会重复）；过滤后仍要 drain 到 END marker。replay 成功完成（返回列表，含空列表）后立即把 `_last_seq` 推进到 `current_seq - 1`，与实时消息 decode 成败解耦，避免同一 gap 被重复 replay。replay 中途 abort（超时、异常、帧格式错误、decode 失败）必须调用 `_replace_replay_socket()` 更换 DEALER socket，防止残留帧污染下一次 replay。replay 使用 DEALER socket（非 REQ），避免 async 下严格一问一答的阻塞问题。
+- **配置优先级**：CLI 显式参数 > `--config yaml` 文件 > dataclass 默认值；`kvcm_base_url` 为必填项（只能由 CLI 或配置文件提供），空值在 `validate()` 阶段直接报错。
+- **启动必需的身份环境变量**：`SPECTRUM_DEPLOYMENT_NAME` 必须为非空唯一部署标识；`KvcmClient.start()` 会校验，缺失/空白直接失败（fatal），因为空值会退化为 `_<block_size>` 这类被无关实例共享的 instance_id，破坏"KVCache 仅在同一 instance_id 内复用"的隔离不变量。
 - **Lint/类型检查禁用规则**：禁止通过内联注释（如 `# noqa`、`# type: ignore`）禁用 ruff 或 mypy 检查。自动生成的代码（如 protobuf 生成文件）应在 `pyproject.toml` 或 `ruff.toml` 中通过 `exclude` / `per-file-ignores` 排除整个文件；只有确实无法通过修改代码解决的极少数情况，才允许使用注释禁用，且必须在注释中说明原因。
 - **Docstring 规范**：通用接口/模块定义（后续要实现某个接口的 placeholder）、抽象基类的 public 方法、复杂实现逻辑必须有 docstring，说明契约、默认行为和关键不变量。简单的内部 helper 和自解释代码不需要。
 - **设计文档落盘目录**：所有 spec / 设计文档写入 `docs/superpowers/specs/`，命名格式为 `YYYY-MM-DD_slug.md`（如 `2026-07-21_engine-node-metadata-rpc.md`）。不要在项目根目录或其他位置创建 `specs/` 目录。
@@ -102,7 +104,7 @@ subscriber 与同机 DashServing "同生同死"：subscriber 未 active 时 Dash
 **核心不变量（修改代码前必须理解）：**
 
 - **引擎健康只由 `GetWorkerStatus.alive` 驱动**：不用 HTTP `/readiness` 做引擎健康判断，不用 DashServing 探针反馈 engine DEAD。
-- **KVCM 是 lossy 旁路，不是门控**：KVCM 不可用不阻塞转发、不影响探针、不改变 subscriber 状态。发送失败丢弃并计数，恢复后自动续传。不要为 KVCM 加 generation/snapshot/condition 等复杂机制。
+- **KVCM 是 lossy 旁路，不是门控（运行期）**：进入 serving 后，KVCM 不可用不阻塞转发、不影响探针、不改变 subscriber 状态。发送失败丢弃并计数，恢复后自动续传。不要为 KVCM 加 generation/snapshot/condition 等复杂机制。**启动期例外（有意设计）**：`_graceful_startup` 第 3 步会等待 KVCM 注册成功（`_wait_kvcm_registered`）才启动 pipeline 并上报 active——启动时需要 KVCM 注册成功才能够启动，避免在从未注册的情况下导流；等待期间 subscriber 保持 starting（readiness 503），并周期性打印 "still waiting for kvcm registration" 警告。不要把这个启动等待当作 bug"修掉"。
 - **seq_id 线性化**：状态上报靠单调递增 seq_id 防乱序。低 seq 的 active 不能复活已接受的 inactive。不要加 session_id 除非需要支持独立重启 subscriber。
 - **HostDown 幂等**：`AllBlocksCleared` 每个 sendable epoch 最多发一次。冷启动失败不发。
 - **同生同死探针语义**：starting → readiness 503 / liveness 200（startup grace）；active → 200/200；inactive/failed/TTL expired → 503/503。
@@ -144,3 +146,4 @@ class SGLangAdapter(AbstractEngineAdapter):
 | 编号 | 内容 | 文件 | 阶段 |
 |---|---|---|---|
 | S-4 | subscriber 不可用恢复策略：kvcm SDK 心跳、TTL、session/checkpoint | kvcm SDK / kvcm 服务端 / subscriber | DONE（heartbeat + TTL + state reporter 已实现；session/checkpoint 见 TODO(independent-restart)） |
+| S-5 | 多 DP（`data_parallel_size > 1`）事件订阅：config 已按 rank 生成 per-rank ZMQ endpoint，但 `VllmIncrementalSource` 目前只订阅 `dp_endpoints[0]`（rank 0），rank 1+ 的 KV 事件不会被转发 | subscriber/engine/vllm/incremental.py | TODO（`subscriber_health_enabled=true` 时已强制 `data_parallel_size == 1`；实现多 DP 前不要放开该限制） |

subscriber/README.md

@@ -10,15 +10,21 @@ uv sync
 
 ## Running
 
-```bash
-# With defaults
-uv run python -m subscriber
+Required at startup:
+
+- `SPECTRUM_DEPLOYMENT_NAME` — unique deployment identity used to build the
+  KVCM `instance_id`; startup fails if missing or blank.
+- KVCM base URL — via `--kvcm-base-url` or the config file; a blank value is
+  rejected by config validation.
 
+```bash
 # With CLI args
-uv run python -m subscriber --zmq-pub-endpoint tcp://localhost:5557 --kvcm-addr 10.0.0.1:50051
+SPECTRUM_DEPLOYMENT_NAME=my-deployment uv run python -m subscriber \
+  --zmq-pub-endpoint tcp://localhost:5557 \
+  --kvcm-base-url spectrum://vs-example:6382
 
-# With config file
-uv run python -m subscriber --config config.yaml
+# With config file (kvcm_base_url can be set in the yaml)
+SPECTRUM_DEPLOYMENT_NAME=my-deployment uv run python -m subscriber --config config.yaml
 ```
 
 ## Development

subscriber/subscriber/config.py

@@ -273,6 +273,8 @@ def validate(self) -> None:
             raise ValueError("kv_event_merge_max_queue_items must be >= 1")
         if not 1 <= self.host_port <= 65535:
             raise ValueError("host_port must be in [1, 65535]")
+        if not self.kvcm_base_url.strip():
+            raise ValueError("kvcm_base_url is required")
         if self.kvcm_heartbeat_interval_s <= 0:
             raise ValueError("kvcm_heartbeat_interval_s must be > 0")
         if self.kvcm_request_timeout_s <= 0:

subscriber/subscriber/engine/vllm/control.py

@@ -203,8 +203,8 @@ async def watch_liveness(self) -> AsyncGenerator[LivenessEvent, None]:
                 consecutive_failures = 0
 
             was_healthy = is_healthy
-            await asyncio.sleep(self._config.engine_health_interval_s)
             yield event
+            await asyncio.sleep(self._config.engine_health_interval_s)
 
     async def fetch_kv_cache_group_metadata(self) -> KvCacheGroupMetadata:
         """Fetch per-group metadata with bounded exponential-backoff retries.

subscriber/subscriber/engine/vllm/incremental.py

@@ -246,6 +246,12 @@ async def subscribe(self) -> AsyncGenerator[EngineEventBatch, None]:
                     )
                     if generation != self._generation:
                         continue
+                    if replay_batches is not None:
+                        # The gap is resolved up to the live message: every
+                        # missing batch still buffered was forwarded and the
+                        # rest is permanently lost. Advance now so a live
+                        # decode failure below cannot replay the same range.
+                        self._last_seq = seq - 1
                     if replay_batches:
                         replay_telemetry.mark("replay_fetch")
                         yield EngineEventBatch(
@@ -454,7 +460,17 @@ async def _replay_missing_batches(
                 return None
             if generation != self._generation:
                 return None
-            if len(frames) != 3:
+            if len(frames) == 4:
+                # vLLM PR #45177 added topic to replay replies. ROUTER
+                # consumes client_id, so DEALER receives
+                # (delimiter, topic, seq_bytes, payload).
+                _, _, replay_seq_bytes, replay_payload = frames
+            elif len(frames) == 3:
+                # Before vLLM PR #45177, replay replies had no topic and
+                # DEALER received (delimiter, seq_bytes, payload).
+                _, replay_seq_bytes, replay_payload = frames
+            else:
+                self._replace_replay_socket(dealer, generation)
                 logger.warning(
                     "dropping malformed kv event replay frames",
                     step="zmq_replay",
@@ -466,7 +482,7 @@ async def _replay_missing_batches(
                     },
                 )
                 return None
-            if frames[1] == _END_SEQ:
+            if replay_seq_bytes == _END_SEQ:
                 if logger.is_debug_enabled():
                     logger.debug(
                         "completed vLLM ZMQ replay",
@@ -479,9 +495,15 @@ async def _replay_missing_batches(
                         },
                     )
                 return replay_batches
-            replay_seq = int.from_bytes(frames[1], "big")
+            replay_seq = int.from_bytes(replay_seq_bytes, "big")
+            # The replay endpoint returns every buffered batch with
+            # seq >= gap_start_seq, including the live batch that exposed the
+            # gap and anything newer; those arrive (or arrived) live, so only
+            # the missing range is forwarded here. Keep draining to END.
+            if replay_seq < gap_start_seq or replay_seq >= current_seq:
+                continue
             batch = self._msgpack_helper.decode(
-                frames[2],
+                replay_payload,
                 step="zmq_replay",
                 tags={
                     "gap_start_seq": gap_start_seq,
@@ -490,6 +512,7 @@ async def _replay_missing_batches(
                 },
             )
             if batch is None:
+                self._replace_replay_socket(dealer, generation)
                 return None
             if logger.is_debug_enabled():
                 logger.debug(

subscriber/subscriber/kvcm/client.py

@@ -89,7 +89,10 @@ def __init__(
         self._supported_mediums = supported_mediums
         self._group_by_idx: dict[int, KvCacheGroupSpec] | None = (
             {spec.group_idx: spec for spec in group_metadata}
-            if group_metadata is not None
+            # An empty metadata tuple is a valid topology; it registers the
+            # same default location spec as the no-metadata path because KVCM
+            # rejects empty location_spec_infos.
+            if group_metadata
             else None
         )
         self._manager_client: AbstractKvCacheManagerClient = (
@@ -131,19 +134,34 @@ def is_registered(self) -> bool:
 
     def _base_url(self) -> str:
         if not self._config.kvcm_base_url:
-            raise ValueError(
-                "kvcm_base_url is required (pass --kvcm-base-url or set "
-                "KV_EVENT_SUBSCRIBER_KVCM_BASE_URL)"
-            )
+            raise ValueError("kvcm_base_url is required")
         return self._config.kvcm_base_url
 
     def _instance_group(self) -> str:
         return self._config.kvcm_instance_group
 
     def _instance_id(self) -> str:
-        deployment = os.environ.get("SPECTRUM_DEPLOYMENT_NAME", "")
+        deployment = self._require_deployment_name()
         return f"{deployment}_{self._effective_block_size()}"
 
+    @staticmethod
+    def _require_deployment_name() -> str:
+        """Return the unique deployment identity, rejecting a blank value.
+
+        A blank SPECTRUM_DEPLOYMENT_NAME would produce a degenerate
+        instance_id such as ``_16`` shared by every replica missing the
+        variable, breaking the cross-instance KVCache isolation invariant.
+        """
+
+        deployment = os.environ.get("SPECTRUM_DEPLOYMENT_NAME", "").strip()
+        if not deployment:
+            raise ValueError(
+                "SPECTRUM_DEPLOYMENT_NAME must be set to a unique deployment "
+                "identity; a blank value would share one KVCM instance_id "
+                "across unrelated engine instances"
+            )
+        return deployment
+
     def _host_ip_port(self) -> str:
         if self._host_ip_port_value is not None:
             return self._host_ip_port_value
@@ -315,6 +333,7 @@ def _report_events_for_batches(
         )
 
     async def start(self) -> None:
+        self._require_deployment_name()
         self._host_ip_port_value = await resolve_host_ip_port(self._config.host_port)
         await self._manager_client.start()
         self._started = True

subscriber/subscriber/kvcm/manager_client.py

@@ -15,6 +15,7 @@
 
 from subscriber import logger
 from subscriber.kvcm.base import AbstractKvCacheManagerClient
+from subscriber.kvcm.errors import KvcmUnavailableError
 from subscriber.kvcm.service_discovery import (
     ServiceDiscovery,
     create_service_discovery,
@@ -151,15 +152,19 @@ async def is_ready(self) -> bool:
         endpoint = self._service_discovery.get_one_endpoint()
         if endpoint is None:
             return False
-        self.base_url = f"http://{endpoint.host}"
-        logger.info(
-            "service discovery recovered manager endpoint",
-            step=_DISCOVERY_STEP,
-            tags={
-                "discovery_type": self._service_discovery.get_type(),
-                "base_url": self.base_url,
-            },
-        )
+        async with self._leader_lock:
+            # Re-check under the lock: a concurrent leader discovery may have
+            # resolved base_url already, and it must not be clobbered.
+            if not self.base_url.startswith(("http://", "https://")):
+                self.base_url = f"http://{endpoint.host}"
+                logger.info(
+                    "service discovery recovered manager endpoint",
+                    step=_DISCOVERY_STEP,
+                    tags={
+                        "discovery_type": self._service_discovery.get_type(),
+                        "base_url": self.base_url,
+                    },
+                )
         return True
 
     # ----- leader discovery internals -----
@@ -424,6 +429,14 @@ async def _request(
                         step=_REQUEST_STEP,
                         tags={"endpoint": endpoint},
                     )
+                    # A leader failover in progress is a transient outage, not
+                    # a report rejection: check_response would raise the
+                    # rejected-report RuntimeError and the batch would be
+                    # dropped permanently instead of retried.
+                    raise KvcmUnavailableError(
+                        f"KVCM leader unavailable for {endpoint}: "
+                        "SERVER_NOT_LEADER and leader discovery exhausted"
+                    )
 
             if check_response:
                 status = payload.get("header", {}).get("status", {})

subscriber/subscriber/main.py

@@ -433,6 +433,7 @@ async def _report_active_until_accepted(self) -> None:
                 return
             except (
                 httpx.TransportError,
+                httpx.HTTPStatusError,
                 TimeoutError,
                 OSError,
             ) as exc:

subscriber/subscriber/metrics/_base.py

@@ -34,12 +34,21 @@ def _dashlog_counter(
 ) -> None:
     # dashlog EASClient::AddCounters silently drops data when OTel is
     # disabled; route counters through Gauge to reach /realtime_metrics.
+    # Limitation: the reset (and idle-key zeroing) is driven lazily by the
+    # next counter that fires, so if every counter goes idle the last window
+    # values remain until any counter fires again.
     global _counter_last_reset_s
     now = time.monotonic()
+    key = (name, tuple(sorted((tags or {}).items())))
     if now - _counter_last_reset_s >= _RESET_INTERVAL_S:
+        stale_keys = [k for k in _counter_accumulators if k != key]
         _counter_accumulators.clear()
         _counter_last_reset_s = now
-    key = (name, tuple(sorted((tags or {}).items())))
+        for stale_name, stale_tags in stale_keys:
+            try:
+                _dashlog_gauge(stale_name, 0.0, tags=dict(stale_tags) or None)
+            except Exception:
+                pass
     accumulated = _counter_accumulators.get(key, 0.0) + value
     _counter_accumulators[key] = accumulated
     try: