Solo's Blog

切换主题

ExecServer并发与集成测试

从 dispatcher lane、真实子进程 harness、断连代理和参数化场景解释 exec-server 并发与集成测试。

基于rust-v0.150.0
CodexRustExecutionTesting
1575
7.9m

ExecServer并发与集成测试 ​

exec-server 的测试不是只调用 handler 方法。单元测试验证 concurrency limit、request span 和 router helper；集成测试则启动真实 exec-server 子进程，通过 WebSocket 发送 JSON-RPC，并用 disconnectable proxy、local/remote 参数化和真实 filesystem/process 资源验证端到端行为。

本文承接ExecServer架构、ExecServer连接与握手和ExecServer进程RPC，面向需要修改 dispatcher 或补集成测试的读者。范围是 request dispatch 与 test harness，不重复各业务 RPC 的实现。读完后，你应能选择单元测试还是进程级集成测试，并写出输入、关键断言和未证明边界。

1. 并发调度 ​

1.1 调度模式 ​

RequestDispatchMode::Inline 直接 await route，保持请求顺序；Concurrent 模式创建 ordinary/control 两个 semaphore lane。initialize 或尚未 initialized 时仍 inline，避免握手状态被并发请求越过。

源码位置：codex-rs/exec-server/src/server/request_dispatcher.rs :: dispatch_request

rust
let Some(RequestLanes { ordinary, control }) = &self.lanes else {
    return task.await;
};
if method == INITIALIZE_METHOD || !self.initialized {
    return task.await;
}
let admission = if matches!(
    method,
    ENVIRONMENT_INFO_METHOD
        | ENVIRONMENT_STATUS_METHOD
        | EXEC_SIGNAL_METHOD
        | EXEC_TERMINATE_METHOD
        | FS_CLOSE_METHOD
) {
    Arc::clone(control)
} else {
    Arc::clone(ordinary)
};

control lane 为健康检查和清理保留容量，避免 ordinary 请求堵塞时无法 signal、terminate 或 close。

1.2 Task集合 ​

并发 request 被放入 JoinSet。连接关闭、route task panic 或 response channel 失败都会转换为 ConnectionClosed；shutdown 会 abort 所有任务并 drain JoinSet。

源码位置：codex-rs/exec-server/src/server/request_dispatcher.rs :: join_next、shutdown

rust
pub(super) async fn join_next(&mut self) -> RequestTaskResult {
    match self.tasks.join_next().await {
        Some(Ok(result)) => result,
        Some(Err(error)) => {
            warn!("exec-server request task failed: {error}");
            RequestTaskResult::ConnectionClosed
        }
        None => RequestTaskResult::Completed,
    }
}

pub(super) async fn shutdown(mut self) {
    self.tasks.abort_all();
    while self.tasks.join_next().await.is_some() {}
}

2. Harness所有权 ​

ExecServerHarness 拥有临时 CODEX_HOME、helper paths、真实 child、WebSocket URL、连接和 request ID 序列。Drop 会 kill child，显式 shutdown 则等待退出。

源码位置：codex-rs/exec-server/tests/common/exec_server.rs :: ExecServerHarness

rust
pub(crate) struct ExecServerHarness {
    codex_home: TempDir,
    _helper_paths: TestCodexHelperPaths,
    child: Child,
    websocket_url: String,
    websocket: WebSocketStream<MaybeTlsStream<TcpStream>>,
    next_request_id: i64,
}

impl Drop for ExecServerHarness {
    fn drop(&mut self) {
        let _ = self.child.start_kill();
    }
}

2.1 启动流程 ​

harness 启动 codex exec-server --listen ws://127.0.0.1:0，从 stdout 读取实际 URL，再带重试连接 WebSocket。测试因此覆盖真实 CLI、listener、connection、dispatcher 和 handler。

源码位置：codex-rs/exec-server/tests/common/exec_server.rs :: exec_server_with_env

rust
let mut child = Command::new(&helper_paths.codex_exe);
child.args(["exec-server", "--listen", "ws://127.0.0.1:0"]);
child.stdin(Stdio::null());
child.stdout(Stdio::piped());
child.env("CODEX_HOME", codex_home.path());
let mut child = child.spawn()?;
let websocket_url = read_listen_url_from_stdout(&mut child).await?;
let (websocket, _) = connect_websocket_when_ready(&websocket_url).await?;

3. 故障注入 ​

3.1 可暂停代理 ​

DisconnectableWebSocketProxy 在 harness 和 server 之间转发字节，测试可以 pause upstream、等待连接阻塞，再 resume。它用于验证 transport disconnect/recovery，而不是模拟业务错误。

源码位置：codex-rs/exec-server/tests/common/exec_server.rs :: DisconnectableWebSocketProxy

rust
pub(crate) struct DisconnectableWebSocketProxy {
    websocket_url: String,
    pause_tx: Option<oneshot::Sender<()>>,
    blocked_connection_rx: Option<oneshot::Receiver<()>>,
    resume_tx: Option<oneshot::Sender<()>>,
    task: JoinHandle<()>,
}

3.2 关闭与超时 ​

next_event_with_timeout 对 WebSocket frame 设置明确 deadline，并同时接受 text/binary JSON；connection close 转成测试错误。测试不会无限挂起。

源码位置：codex-rs/exec-server/tests/common/exec_server.rs :: next_event_with_timeout

rust
let frame = timeout(timeout_duration, self.websocket.next())
    .await
    .map_err(|_| anyhow!("timed out waiting for exec-server websocket event"))?
    .ok_or_else(|| anyhow!("exec-server websocket closed"))??;

4. 测试层次 ​

层次输入关键断言未证明
dispatcher单元mode、method、tracelimit、lane、span真实transport
handshake集成raw JSON-RPC顺序初始化错误顺序业务handler结果
process集成ExecParams、stdin、signalseq、exit、closed所有平台sandbox
filesystem集成PathUri、sandbox context内容、逃逸拒绝外部文件系统
disconnect集成proxy pause/resumesession/process恢复公网质量

这个矩阵用于选择最小测试层；不能用一个 process happy path 替代 handshake、dispatcher 或 disconnect 断言。

5. 代表性验证 ​

5.1 握手顺序 ​

exec_server_rejects_pipelined_requests_before_initialized 在 concurrent 模式下先发送 environment/info，再发送 initialize 和 initialized，断言响应仍保持 wire-order 的初始化错误。它证明并发调度不会绕过握手门槛。

源码位置：codex-rs/exec-server/tests/initialize.rs :: exec_server_rejects_pipelined_requests_before_initialized

text
cd codex-rs
cargo test -p codex-exec-server --test initialize exec_server_rejects_pipelined_requests_before_initialized -- --test-threads=1

5.2 并发与恢复 ​

process integration 在 --concurrent-requests 32 下验证请求并行和 control 操作；disconnect recovery 测试通过 proxy 暂停连接，再恢复并继续读取同一 process。它证明队列、session resume 和 replay 协作，不证明无限并发或无界断线时间。

源码位置：codex-rs/exec-server/tests/process.rs、codex-rs/exec-server/tests/exec_process.rs :: remote_exec_process_recovers_after_transport_disconnect

text
cd codex-rs
cargo test -p codex-exec-server --test process -- --test-threads=1
cargo test -p codex-exec-server --test exec_process remote_exec_process_recovers_after_transport_disconnect -- --test-threads=1
cargo test -p codex-exec-server --lib server::request_dispatcher::tests -- --test-threads=1

6. 源码排查 ​

text
rg -n "RequestDispatchMode|RequestLanes|JoinSet|control" codex-rs/exec-server/src/server/request_dispatcher.rs
rg -n "ExecServerHarness|disconnectable_websocket_proxy|next_event_with_timeout" codex-rs/exec-server/tests/common/exec_server.rs
rg -n "concurrent-requests|pipelined|recovers_after_transport_disconnect" codex-rs/exec-server/tests

本篇的学习闭环是：用单元测试证明调度策略，用真实 child harness 证明 wire 与资源生命周期，用代理注入断连，再以 local/remote 参数化验证共同契约。下一篇Apply Patch语言与语法进入 Apply Patch 子系列。