执行摘要

PR #52671 修正了 vLLM Rust 前端 engine-core client 中 call_utility 的多引擎扇出语义：由"遇错即返"（fail-fast）改为"等待所有引擎收敛后再返回错误"，使调用方能够安全补偿部分已应用的 mutation；同时引入 RAII 守卫 UtilityCallGuard 统一注销 utility waiter，并把取消安全对齐到普通生成请求 call。改动仅限 Rust 前端 vllm-engine-core-client crate，wire format 与 Python 协议完全不变，附带 3 个回归测试，98 项测试全部通过。

功能与动机

PR body 明确提出："Make multi-engine utility fanout wait for every engine outcome before returning an error"，并强调要让调用方"compensate partially applied mutations only after every engine call has finished"。

根本问题在于旧实现的三阶段设计：phase 2 用 try_join_all 发送、phase 3 用 try_join_all 等待响应，都是 fail-fast。一旦某个 engine 先失败（例如返回 failure_message: "boom"），调用方立刻拿到错误，但其余 engine 可能仍在执行同一个 mutation（如 add_lora、pause/resume），调用方无法判断哪些引擎已应用变更，补偿操作既不安全也不完整。此外，旧代码需要手动回滚已分配的 call_id，准备失败、发送失败、取消三个路径各有清理逻辑，存在 waiter 泄漏风险。

实现拆解

1. 核心语义改造（client.rs call_utility）：把"发送 + 等待响应 + 类型化解码"合并为每个 engine 的单个 future，统一交给 join_all 驱动；try_join_all 被移除。join_all 保证所有 future 都执行完，最后 outcomes.into_iter().collect() 把 Vec<Result<T>> 折叠为 Result<Vec<T>>。任一 engine 失败时，其余引擎的变更已落定，调用方补偿才安全。

2. 清理路径收敛为 RAII：新增局部结构体 UtilityCallGuard，Drop 中 drain 掉 call_ids 并调用 unregister_utility_calls。准备失败（? 提前返回）、发送失败、future 被 cancel 三种情况全部走同一条清理路径，旧代码分散的手动回滚分支被移除。

3. call 取消安全对齐（BugenZhao 追加 commit）：call 改为在发送前先构造 EngineCoreOutputStream（client.rs），并加注释说明这样可保证 cancel 时经 stream 的 Drop 清理请求注册；配套新增 #[cfg(test)] 的 ClientInner::pending_utility_call_count（imp.rs）与 UtilityRegistry::len（state.rs），用于测试断言内部注册表状态。

4. 测试配套（tests/client.rs）：原 call_utility_failure_message_surfaces_as_error 重写为 call_utility_waits_for_all_engines_before_returning_error（双 mock engine：engine0 立即回 failure、engine1 挂起，断言 pending_utility_call_count 收敛为 1 且 call.is_finished() 为 false，释放后错误才透出）；新增 cancelled_utility_call_unregisters_waiter 验证取消后注册表清零且 Arc::try_unwrap 成功；保留 dispatcher_failure_propagates_to_waiting_utility_calls。

rust/src/engine-core-client/src/client.rs

核心改造文件：call_utility 由 fail-fast 改为 join_all 全量收敛，引入 UtilityCallGuard RAII 清理；call 改为先构造输出流再做发送以确保取消安全。

/// Call a typed utility method on all connected engines, returning one
/// decoded result per connected engine if all calls succeed.
///
/// 关键语义：与旧的 fail-fast（`try_join_all`）不同，这里会等待所有
/// engine 都返回结果（成功或失败）后才对外暴露错误，这样调用方只有在
/// 每个引擎的变更都落定之后，才能安全地补偿部分已应用的 mutation。
pub async fn call_utility<T, A>(&self, method: &str, args: A) -> Result<Vec<T>>
where
    T: serde::de::DeserializeOwned,
    A: serde::Serialize + std::fmt::Debug,
{
    // RAII 守卫：无论函数提前返回、发送中途失败，还是整个 future 被
    // cancel（drop），`Drop` 都会统一把已分配的 call_id 从注册表中删除，
    // 避免 waiter 泄漏到 shutdown。旧代码需要手写多个回滚分支，容易漏。
    struct UtilityCallGuard<'a> {
        inner: &'a ClientInner,
        call_ids: Vec<u64>,
    }
​
    impl Drop for UtilityCallGuard<'_> {
        fn drop(&mut self) {
            self.inner.unregister_utility_calls(self.call_ids.drain(..));
        }
    }
​
    let mut call_guard = UtilityCallGuard {
        inner: self.inner.as_ref(),
        call_ids: Vec::with_capacity(self.engines.len()),
    };
​
    // 为每个 connected engine 分配唯一的 call_id 并注册响应 waiter，
    // 同时按 Python 侧 `(client_index, call_id, method_name, args)`
    // 的元组契约构建请求 payload。
    let mut prepared_calls = Vec::with_capacity(self.engines.len());
    for engine in &self.engines {
        let (call_id, rx) = self.inner.allocate_and_register_utility_call()?;
        call_guard.call_ids.push(call_id);
        let request = EngineCoreUtilityRequest::new(
            self.config.client_index,
            call_id,
            method,
            &args,
        )?;
        prepared_calls.push((&engine.engine_id, call_id, rx, request));
    }
​
    // 并发发送 + 等待响应。`join_all` 不会像 `try_join_all` 那样在第一个
    // 错误出现时取消其余 future，而是等所有结果收敛后再统一处理。
    let outcomes = join_all(prepared_calls.into_iter().map(
        |(engine_id, call_id, rx, request)| async move {
            self.inner
                .send_to_engine(engine_id, EngineCoreRequestType::Utility, &request)
                .await?;
            rx.await
                .map_err(|_| Error::UtilityCallClosed {
                    method: method.to_string(),
                    call_id,
                })??
                .into_typed_result(method)
        },
    ))
    .await;
​
    // `Vec<Result<T>>` 折叠为 `Result<Vec<T>>`，结果列表保持 engine 顺序；
    // 此时所有 engine 均已返回，调用方可以安全地执行补偿逻辑。
    outcomes.into_iter().collect()
}

// `call` 侧与 `call_utility` 对齐取消安全：先构造输出流，再真正向 engine
// 发送。这样一旦在发送过程中 future 被 cancel，stream 被 drop 时仍会走
// `EngineCoreOutputStream` 的 `Drop` 逻辑把请求注册回滚掉，不会留下
// 悬挂的注册项。
pub async fn call(&self, mut req: EngineCoreRequest) -> Result<EngineCoreOutputStream> {
    req.client_index = self.config.client_index;
    req.validate()?;
​
    let (engine_id, rx) =
        self.inner.register_request(req.request_id.clone(), lora_name, data_parallel_rank)?;
​
    let stream = EngineCoreOutputStream::new(
        req.request_id.clone(),
        engine_id.engine_index().unwrap_or(0),
        self.abort_tx.clone(),
        rx,
    );
​
    let result: Result<()> = async {
        // 协调器存在时，用协调器快照覆盖当前 wave，并在引擎未运行时
        // 通知第一个请求，触发引擎启动流程。
        if let Some(coordinator) = self.coordinator.as_ref() {
            let snapshot = coordinator.snapshot();
            req.current_wave = snapshot.current_wave;
            if !snapshot.engines_running {
                coordinator.notify_first_request(engine_id.clone())?;
            }
        }
        self.inner.send_request_to_engine(&engine_id, req).await?;
        Ok(())
    }
    .await;
​
    // 发送失败则回滚注册，并丢弃已构造的 stream（其 `Drop` 兜底清理）。
    if let Err(error) = result {
        self.inner.rollback_request(&req.request_id);
        return Err(error);
    }
​
    Ok(stream)
}

rust/src/engine-core-client/src/tests/client.rs

回归测试核心文件：把原单引擎失败测试重写为双引擎等待测试，新增取消清理测试，验证本 PR 的两个关键行为。

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn cancelled_utility_call_unregisters_waiter() {
    init_tracing();
    let ipc = IpcNamespace::new().unwrap();
    let handshake_address = ipc.handshake_endpoint();
    let (received_tx, received_rx) = oneshot::channel();
​
    // mock engine：收到 utility 请求后只上报 " 已收到 "，然后永远挂起，
    // 模拟一个迟迟不给响应的远端。
    let (_shutdown, engine_task) = spawn_mock_engine_task(
        handshake_address.clone(),
        vec![0x00, 0x00],
        |dealer, _push| {
            Box::pin(async move {
                let _utility = recv_engine_message(dealer).await;
                let _ = received_tx.send(());
                std::future::pending::<()>().await;
            })
        },
    );
​
    let client = std::sync::Arc::new(connect_client_with_ipc(/* ... */).await);
​
    // 发起 utility 调用后，注册表中应当立即出现 1 个 pending waiter。
    let call_client = client.clone();
    let call =
        tokio::spawn(async move { call_client.call_utility::<bool, _>("add_lora", ()).await });
    received_rx.await.unwrap();
    assert_eq!(client.pending_utility_call_count(), 1);
​
    // 取消调用：注册表中不再有残留 waiter；同时 Arc 还能被 `try_unwrap`，
    // 说明没有任何任务在结束后仍持有 client。
    call.abort();
    assert!(call.await.unwrap_err().is_cancelled());
    assert_eq!(client.pending_utility_call_count(), 0);
​
    engine_task.abort();
    let client = std::sync::Arc::try_unwrap(client)
        .unwrap_or_else(|_| panic!("utility task retained client after cancellation"));
    client.shutdown().await.unwrap();
}

评论区精华

BugenZhao（review approval）："Thanks and this is a good catch! I've pushed some more commits to - simplify the impl a bit - also align the cancellation-safety behavior with the normal generation call"

这段点评点出了本 PR 的两个关键设计决策：一是用 RAII 守卫把多条清理分支收敛成单一 Drop 路径；二是把取消安全从 utility 扇出推广到普通生成请求 call，形成一致的"先建资源、再发请求"模式。测试里的 Arc::try_unwrap 断言也值得注意——它验证没有任何异步任务在完成后仍持有 client，是排查未来泄漏的哨兵。

风险与影响

• 错误返回延迟拉长：即使某个 engine 已失败，也要等全部引擎返回才报错；若引擎挂起且无超时兜底，call_utility 可能长时间阻塞。这是补偿安全与时效性的刻意取舍。
• 级联语义变化：call_utility_consensus 与 collective_rpc 委托 call_utility，失败语义统一变为全量收敛，所有 utility 调用方都受影响。
• 取消清理依赖 Drop 兜底：call 的取消安全依赖 EngineCoreOutputStream::Drop 的回滚实现，未来改动 stream 语义时需保持该契约。
• 影响面：仅 Rust 前端 client crate，Python 协议与推理输出不受影响；多 engine（DP）部署下的 LoRA 增删、pause/resume 等操作获得确定的补偿语义，同时减少了 waiter 泄漏。

关联脉络

• PR #52575（[Rust Frontend] Simplify data-parallel size ownership）：同属 Rust engine-core client 的多引擎数据并行改造线，改动同一文件 client.rs；本 PR 的 utility 全量收敛语义正是为多 engine（DP）部署的补偿安全服务。
• PR #52144（[Test] Add pause/resume E2E tests）：pause/resume 通过 utility 协议（PauseMode）在多引擎上执行，本 PR 保证所有引擎收敛后再报错，正好补齐这类状态变更的补偿安全，两者共同完善多引擎生命周期管理。
• 从近期历史看，Rust frontend 正在经历密集的"多引擎语义收敛"改造（DP 大小归属、utility 扇出等待、取消安全），本 PR 是其中关键一环，为后续 utility 协议扩展提供了可复用的错误与清理模型。