socks5: send empty keepalive msg to avoid triggering MIX_TTL during long downloads (#3364)

* socks5: send empty keepalive msg to avoid triggering MIX_TTL during long downloads

* rustfmt

* reset timer after each normal send

common/socks5/proxy-helpers/src/proxy_runner/inbound.rs

@@ -4,6 +4,7 @@
 use super::MixProxySender;
 use super::SHUTDOWN_TIMEOUT;
 use crate::available_reader::AvailableReader;
+use crate::proxy_runner::KEEPALIVE_INTERVAL;
 use bytes::Bytes;
 use futures::FutureExt;
 use futures::StreamExt;
@@ -35,6 +36,23 @@ async fn send_empty_close<F, S>(
         .expect("BatchRealMessageReceiver has stopped receiving!");
 }
 
+async fn send_empty_keepalive<F, S>(
+    connection_id: ConnectionId,
+    message_sender: &mut OrderedMessageSender,
+    mix_sender: &MixProxySender<S>,
+    adapter_fn: F,
+) where
+    F: Fn(ConnectionId, Vec<u8>, bool) -> S,
+    S: Debug,
+{
+    log::trace!("Sending keepalive for connection: {connection_id}");
+    let ordered_msg = message_sender.wrap_message(Vec::new()).into_bytes();
+    mix_sender
+        .send(adapter_fn(connection_id, ordered_msg, false))
+        .await
+        .expect("BatchRealMessageReceiver has stopped receiving!");
+}
+
 #[allow(clippy::too_many_arguments)]
 async fn deal_with_data<F, S>(
     read_data: Option<io::Result<Bytes>>,
@@ -185,6 +203,8 @@ where
 
     tokio::pin!(shutdown_future);
 
+    let mut keepalive_timer = tokio::time::interval(KEEPALIVE_INTERVAL);
+
     loop {
         select! {
             read_data = &mut available_reader.next() => {
@@ -200,6 +220,10 @@ where
                 ).await {
                     break
                 }
+                keepalive_timer.reset();
+            }
+            _ = keepalive_timer.tick() => {
+                send_empty_keepalive(connection_id, &mut message_sender, &mix_sender, &adapter_fn).await;
             }
             _ = &mut shutdown_future => {
                 debug!(

common/socks5/proxy-helpers/src/proxy_runner/mod.rs

@@ -15,6 +15,10 @@ mod outbound;
 // TODO: make this configurable
 const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30);
 
+// Send empty keepalive messages regurarly to keep the connection alive. This should be smaller
+// than [`MIX_TTL`].
+const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(60);
+
 #[derive(Debug)]
 pub struct ProxyMessage {
     pub data: Vec<u8>,