fix: flush DataChannelCmd::StartForward* commands (#316)

Without flushing this may sit in a kernel buffer and we won't
know if the channel is still alive. This is particularly problematic
for the TCP connection pool.

Cargo.lock

@@ -458,9 +458,9 @@ dependencies = [
 
 [[package]]
 name = "curve25519-dalek"
-version = "4.1.1"
+version = "4.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e89b8c6a2e4b1f45971ad09761aafb85514a84744b67a95e32c3cc1352d1f65c"
+checksum = "0a677b8922c94e01bdbb12126b0bc852f00447528dee1782229af9c720c3f348"
 dependencies = [
  "cfg-if",
  "cpufeatures",
@@ -1245,6 +1245,15 @@ version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
 
+[[package]]
+name = "openssl-src"
+version = "300.2.3+3.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5cff92b6f71555b61bb9315f7c64da3ca43d87531622120fea0195fc761b4843"
+dependencies = [
+ "cc",
+]
+
 [[package]]
 name = "openssl-sys"
 version = "0.9.93"
@@ -1253,6 +1262,7 @@ checksum = "db4d56a4c0478783083cfafcc42493dd4a981d41669da64b4572a2a089b51b1d"
 dependencies = [
  "cc",
  "libc",
+ "openssl-src",
  "pkg-config",
  "vcpkg",
 ]
@@ -1496,6 +1506,7 @@ dependencies = [
  "hex",
  "lazy_static",
  "notify",
+ "openssl",
  "rand",
  "serde",
  "sha2",

src/helper.rs

@@ -1,8 +1,9 @@
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, Context, Result};
 use async_http_proxy::{http_connect_tokio, http_connect_tokio_with_basic_auth};
 use backoff::{backoff::Backoff, Notify};
 use socket2::{SockRef, TcpKeepalive};
 use std::{future::Future, net::SocketAddr, time::Duration};
+use tokio::io::{AsyncWrite, AsyncWriteExt};
 use tokio::{
     net::{lookup_host, TcpStream, ToSocketAddrs, UdpSocket},
     sync::broadcast,
@@ -144,3 +145,14 @@ where
         }
     }
 }
+
+pub async fn write_and_flush<T>(conn: &mut T, data: &[u8]) -> Result<()>
+where
+    T: AsyncWrite + Unpin,
+{
+    conn.write_all(data)
+        .await
+        .with_context(|| "Failed to write data")?;
+    conn.flush().await.with_context(|| "Failed to flush data")?;
+    Ok(())
+}

src/server.rs

@@ -1,7 +1,7 @@
 use crate::config::{Config, ServerConfig, ServerServiceConfig, ServiceType, TransportType};
 use crate::config_watcher::{ConfigChange, ServerServiceChange};
 use crate::constants::{listen_backoff, UDP_BUFFER_SIZE};
-use crate::helper::retry_notify_with_deadline;
+use crate::helper::{retry_notify_with_deadline, write_and_flush};
 use crate::multi_map::MultiMap;
 use crate::protocol::Hello::{ControlChannelHello, DataChannelHello};
 use crate::protocol::{
@@ -498,14 +498,9 @@ struct ControlChannel<T: Transport> {
 
 impl<T: Transport> ControlChannel<T> {
     async fn write_and_flush(&mut self, data: &[u8]) -> Result<()> {
-        self.conn
-            .write_all(data)
+        write_and_flush(&mut self.conn, data)
             .await
             .with_context(|| "Failed to write control cmds")?;
-        self.conn
-            .flush()
-            .await
-            .with_context(|| "Failed to flush control cmds")?;
         Ok(())
     }
     // Run a control channel
@@ -640,7 +635,7 @@ async fn run_tcp_connection_pool<T: Transport>(
     'pool: while let Some(mut visitor) = visitor_rx.recv().await {
         loop {
             if let Some(mut ch) = data_ch_rx.recv().await {
-                if ch.write_all(&cmd).await.is_ok() {
+                if write_and_flush(&mut ch, &cmd).await.is_ok() {
                     tokio::spawn(async move {
                         let _ = copy_bidirectional(&mut ch, &mut visitor).await;
                     });
@@ -690,7 +685,7 @@ async fn run_udp_connection_pool<T: Transport>(
         .recv()
         .await
         .ok_or_else(|| anyhow!("No available data channels"))?;
-    conn.write_all(&cmd).await?;
+    write_and_flush(&mut conn, &cmd).await?;
 
     let mut buf = [0u8; UDP_BUFFER_SIZE];
     loop {