使用tokio::io::copy_bidirectional代理TCP连接时遇os error 107

我需要基于socket2::Socket实现TCP连接代理，将其转换为tokio::net::TcpStream后，用Tokio的双向复制功能完成数据转发。整体功能看似正常——用curl发起HTTPS请求时，TLS握手能完成且能收到响应，但会触发os error 107 (Transport endpoint is not connected)错误，且预期的"bytes copied"日志从未出现。

初始实现代码 #

let std_stream: std::net::TcpStream = socket.into();
std_stream.set_nonblocking(true)?;
let mut inbound = TcpStream::from_std(std_stream)?;
debug!("created inbound tcp stream from socket");

tokio::spawn(async move {
    match TcpStream::connect(&dest_addr).await {
        Ok(mut outbound) => {
            debug!("connected to {dest_addr}");

            match tokio::io::copy_bidirectional(&mut inbound, &mut outbound).await {
                Ok((nsent, nrcvd)) => {
                    debug!(sent_bytes = nsent, rcvd_bytes = nrcvd, "bytes copied");
                    return Ok(());
                }

                Err(err) => {
                    error!(
                        "💥️ — proxying for address {} failed: {}",
                        &dest_addr, err
                    );
                    return Err(err);
                }
            }
        }
        Err(err) => {
            error!(
                "unable to connect to downstream host {:?}: {err}",
                dest_addr
            );

            return Err(err);
        }
    }
});

错误日志 #

{"timestamp":"2024-04-15T17:33:56.042465Z","level":"DEBUG","fields":{"message":"created inbound tcp stream from socket"},"target":"tfw"}
{"timestamp":"2024-04-15T17:33:56.080270Z","level":"DEBUG","fields":{"message":"connected to 104.244.42.129:443"},"target":"tfw"}
{"timestamp":"2024-04-15T17:33:56.342300Z","level":"ERROR","fields":{"message":"💥️ — proxying for address 104.244.42.129:443 failed: Transport endpoint is not connected (os error 107)"},"target":"tfw"}

排查过程更新 #

4/16/24 10:00 更新：自行实现双向复制逻辑 #

我替换了Tokio内置的copy_bidirectional，改用自定义的双向复制逻辑：

/// Runs the actual proxying of a socket.
async fn transfer(mut incoming: TcpStream, mut outbound: TcpStream) -> Result<(), std::io::Error> {
    // Split the streams into read and write halves.
    let (mut read_inbound, mut write_inbound) = incoming.split();
    let (mut read_outbound, mut write_outbound) = outbound.split();

    // Forward client data to server
    let client_to_server = async {
        tokio::io::copy(&mut read_inbound, &mut write_outbound)
            .await
            .inspect_err(|e| {
                error!(
                    error = e.to_string(),
                    "error copying read_inbound to write_outbound!"
                )
            })?;

        debug!("finished copying read_inbound to write_outbound, shutting down write_outbound");
        write_outbound
            .shutdown()
            .await
            .inspect_err(|e| error!(error = e.to_string(), "error shutting down write_outbound!"))
    };

    // Forward server data to client
    let server_to_client = async {
        tokio::io::copy(&mut read_outbound, &mut write_inbound)
            .await
            .inspect_err(|e| {
                error!(
                    error = e.to_string(),
                    "error copying read_outbound to write_inbound!"
                )
            })?;

        debug!("finished copying read_outbound to write_inbound, shutting down write_inbound");
        write_inbound
            .shutdown()
            .await
            .inspect_err(|e| error!(error = e.to_string(), "error shutting down write_inbound!"))
    };

    // Poll both tasks.
    tokio::try_join!(client_to_server, server_to_client)?;

    Ok(())
}

测试后仍会触发错误，但日志显示错误发生在所有数据复制完成后、关闭write_inbound时，关闭write_outbound则无异常。我推测是curl客户端提前关闭了连接，导致inbound流已处于关闭状态，此时手动执行关闭操作就会触发错误。

4/16/24 10:20 更新：移除write_outbound关闭操作后的验证 #

我移除了write_outbound的关闭逻辑，测试是否关闭一个流会影响另一个，但错误依然在关闭write_inbound时触发，进一步确认是客户端已提前关闭该连接，导致后续手动关闭操作报错。

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