fix(ipc): PORT 文件锚定 exe 目录且不再被多实例覆盖,读写加超时
问题: 1. PORT 文件用相对路径,读写位置随 CWD 变化。 2. 所有实例(显式端口、端口冲突退避到随机端口的)都写同一个 PORT 文件,多实例互相覆盖,客户端可能连错实例;写入失败还会 杀死整个 IPC 服务。 3. 客户端读响应、服务端读请求都没有超时,对端异常时永久挂起。 修复: - PORT 文件锚定 current_exe 目录。 - 仅默认端口实例写 PORT 文件,写入失败降级为告警。 - 服务端读请求超时 10s,客户端读响应超时 5s。 测试:test_port_file_path_is_absolute 验证路径为绝对路径。
This commit is contained in:
@@ -39,7 +39,11 @@ pub async fn run_client(cmd: IpcCmd, port: Option<u16>) -> 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
|
||||
|
||||
+16
-1
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
+18
-3
@@ -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<u16>, 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?;
|
||||
|
||||
Reference in New Issue
Block a user