Rust Tokio Unix Socket服务端广播消息不全测试不通过问题

• 核心逻辑错误：handle函数每次循环都会重新调用tx.subscribe()创建新的广播订阅者，tokio broadcast通道的新订阅者只会从创建时刻的消息偏移位置开始接收数据，之前已发送的消息会被直接跳过，因此客户端最多只能收到1条消息，和日志中所有客户端仅拿到1的现象完全吻合。
• 竞态条件：测试流程中启动feed消息生产任务后立刻发起客户端连接，echo命令执行速度极快，4条测试消息在部分客户端还未完成连接、未完成订阅时就已经全部发送到通道，晚到的订阅者无法获取订阅前的历史消息。
• 终止时机过早：feed任务发完所有消息后立刻触发abort通知，服务端直接终止监听循环、断开客户端连接，此时写入socket缓冲区的消息还未完成传输，也没有做可靠flush，剩余消息直接被丢弃。
• 命令参数错误：feed中调用echo时传入的参数自带\n换行符，而echo默认会给每个参数追加换行，导致输出多余空行，日志中出现的空[Server] 打印就是该问题导致。

1. 修正客户端连接处理逻辑 #

连接建立时仅创建一次订阅者，循环复用该订阅者接收消息，写入数据后及时flush，同时处理通道消息堆积的Lagged错误避免意外断开连接：

async fn handle(mut stream: UnixStream, tx: Sender<String>, abort: Arc<Notify>) {
    // 连接建立时仅订阅一次，禁止在循环内重复创建订阅者
    let mut rx = tx.subscribe();
    loop {
        tokio::select! {
            _ = abort.notified() => break,
            result = rx.recv() => match result {
                Ok(output) => {
                    stream.write_all(output.as_bytes()).await.unwrap();
                    stream.write(b"\n").await.unwrap();
                    // 写入后立即flush，避免消息留在缓冲区未发送
                    stream.flush().await.unwrap();
                }
                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
                    // 消息堆积丢包时跳过丢失的消息，继续接收后续数据
                    continue;
                }
                Err(e) => {
                    println!("[Server] Client connection error: {e}");
                    break;
                }
            }
        }
    }
    let _ = stream.shutdown().await;
}

2. 修正测试流程的竞态问题 #

先发起所有客户端连接，等待所有客户端完成订阅后再启动feed任务发消息，消息发送完成后预留足够传输时间再终止服务：

#[tokio::test]
async fn test_server() -> Result<()> {
    let mut server = Server::new("/tmp/testsock.socket");
    start(&mut server).await?;

    let address = server.address.clone();
    // 先建立所有客户端连接
    let mut client_handles = vec![];
    for name in ["Alpha", "Beta", "Delta", "Gamma"] {
        client_handles.push(tokio::spawn(connect(address.clone(), name.into())));
    }
    // 等待所有客户端完成连接和订阅，生产环境可使用同步屏障替代sleep更可靠
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;

    // 所有客户端就绪后再启动消息生产任务
    let feed_handle = feed(server.tx.clone(), server.abort.clone()).unwrap();
    feed_handle.await??;

    // 等待消息全部传输完成后再终止服务
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    server.abort.notify_waiters();
    server.handle.unwrap().await??;

    // 校验每个客户端都收到全部4条消息
    for (idx, handle) in client_handles.into_iter().enumerate() {
        let messages = handle.await?;
        assert_eq!(messages.len(), 4, "Client {idx} did not receive all messages");
    }

    Ok(())
}

3. 修正feed任务的命令参数 #

去掉echo参数中自带的换行符，避免产生多余空行：

let mut child = Command::new("echo")
    .args(&["1", "2", "3", "4"])
    .stdout(Stdio::piped())
    .stderr(Stdio::null())
    .stdin(Stdio::null())
    .spawn()?;

重新运行测试，所有客户端均可完整收到4条广播消息，测试断言全部通过，日志输出符合预期。

内容的提问来源于stack exchange，提问作者jam2020