diff --git a/vnt-ipc/src/client.rs b/vnt-ipc/src/client.rs index 13bfdd1..9c46aa9 100644 --- a/vnt-ipc/src/client.rs +++ b/vnt-ipc/src/client.rs @@ -39,7 +39,11 @@ pub async fn run_client(cmd: IpcCmd, port: Option) -> anyhow::Result<()> { framed .send(IpcRequest { ipc_cmd: Some(cmd) }.encode_to_vec().into()) .await?; - let response = framed.next().await.context("Unexpected end of stream")??; + // 读响应加超时:服务端异常不回复时客户端不能永久挂起 + let response = tokio::time::timeout(Duration::from_secs(5), framed.next()) + .await + .context("Response timed out")? + .context("Unexpected end of stream")??; let response = IpcResponse::decode(response).context("decode response error")?; match response .response_payload diff --git a/vnt-ipc/src/lib.rs b/vnt-ipc/src/lib.rs index e9fc7aa..e941e17 100644 --- a/vnt-ipc/src/lib.rs +++ b/vnt-ipc/src/lib.rs @@ -8,5 +8,20 @@ const DEFAULT_PORT: u16 = 11233; const PORT_FILE: &str = "PORT"; fn get_port_file_path() -> std::path::PathBuf { - std::path::PathBuf::from(PORT_FILE) + // 锚定可执行文件目录,相对路径会随 CWD 变化导致读不到 PORT 文件 + std::env::current_exe() + .ok() + .and_then(|p| p.parent().map(|dir| dir.join(PORT_FILE))) + .unwrap_or_else(|| std::path::PathBuf::from(PORT_FILE)) +} + +#[cfg(test)] +mod tests { + /// PORT 文件路径必须锚定到可执行文件目录(绝对路径) + #[test] + fn test_port_file_path_is_absolute() { + let path = super::get_port_file_path(); + assert!(path.is_absolute(), "path should be absolute: {path:?}"); + assert_eq!(path.file_name().unwrap(), super::PORT_FILE); + } } diff --git a/vnt-ipc/src/server.rs b/vnt-ipc/src/server.rs index bfd3018..806ee3a 100644 --- a/vnt-ipc/src/server.rs +++ b/vnt-ipc/src/server.rs @@ -11,6 +11,7 @@ use futures::{SinkExt, StreamExt}; use prost::Message; use std::fs; use std::net::Ipv4Addr; +use std::time::Duration; use tokio::io::{self}; use tokio::net::{TcpListener, TcpStream}; use tokio_util::codec::{Framed, LengthDelimitedCodec}; @@ -19,7 +20,14 @@ use vnt_core::api::VntApi; async fn handle_connection(stream: TcpStream, vnt_api: VntApi) -> anyhow::Result<()> { let mut framed = Framed::new(stream, LengthDelimitedCodec::new()); - if let Some(Ok(message)) = framed.next().await { + // 读请求加超时:空闲连接不能永久占用任务 + let message = match tokio::time::timeout(Duration::from_secs(10), framed.next()).await { + Ok(Some(Ok(message))) => message, + Ok(Some(Err(e))) => return Err(e.into()), + Ok(None) => bail!("connection closed without request"), + Err(_) => bail!("read request timed out"), + }; + { let request = IpcRequest::decode(message.as_ref())?; let Some(cmd) = request.ipc_cmd else { bail!("Received an IpcRequest but it was None"); @@ -178,8 +186,15 @@ pub async fn run_server(bind_port: Option, vnt_api: VntApi) -> anyhow::Resu log::info!("IPC Listening on {}", bound_addr); let actual_port = bound_addr.port(); - let path = get_port_file_path(); - fs::write(&path, actual_port.to_string())?; + // 只有默认端口的实例才写 PORT 文件:显式指定端口的实例、以及端口 + // 冲突退避到随机端口的实例都不写,避免多实例互相覆盖导致客户端 + // 连错实例;写入失败不影响服务本身 + if bind_port.is_none() && actual_port == DEFAULT_PORT { + let path = get_port_file_path(); + if let Err(e) = fs::write(&path, actual_port.to_string()) { + log::warn!("write PORT file failed: {e:?}"); + } + } loop { let (stream, peer_addr) = listener.accept().await?;