Made idle timeout configurable for local HTTP clients

changelog.d/+local-http-connection-timeout.added.md

@@ -0,0 +1 @@
+Added an option to configure timeout for idle local HTTP connections (`experimental.idle_local_http_connection_timeout`).

mirrord-schema.json

@@ -823,6 +823,16 @@
             "null"
           ]
         },
+        "idle_local_http_connection_timeout": {
+          "title": "_experimental_ idle_local_http_connection_timeout {#experimental-idle_local_http_connection_timeout}",
+          "description": "Sets a timeout for idle local HTTP connections (in milliseconds).\n\nHTTP requests stolen with a filter are delivered to the local application from a HTTP connection made from the local machine. Once a request is delivered, the connection is cached for some time, so that it can be reused to deliver the next request.\n\nThis timeout determines for how long such connections are cached.\n\nSet to 0 to disable caching local HTTP connections (connections will be dropped as soon as the request is delivered).",
+          "type": [
+            "integer",
+            "null"
+          ],
+          "format": "uint64",
+          "minimum": 0.0
+        },
         "readlink": {
           "title": "_experimental_ readlink {#experimental-readlink}",
           "description": "DEPRECATED, WILL BE REMOVED",

mirrord/cli/src/internal_proxy.rs

@@ -106,6 +106,7 @@ pub(crate) async fn proxy(
         agent_conn,
         listener,
         config.experimental.readonly_file_buffer,
+        Duration::from_millis(config.experimental.idle_local_http_connection_timeout),
     )
     .run(first_connection_timeout, consecutive_connection_timeout)
     .await

mirrord/cli/src/main.rs

@@ -525,6 +525,7 @@ async fn port_forward(args: &PortForwardArgs, watch: drain::Watch) -> CliResult<
                     connection_2,
                     rev_port_mappings,
                     config.feature.network.incoming,
+                    Duration::from_millis(config.experimental.idle_local_http_connection_timeout),
                 )
                 .await?;
                 port_forward.run().await.map_err(|error| error.into())

mirrord/cli/src/port_forward.rs

@@ -440,10 +440,15 @@ impl ReversePortForwarder {
         mut agent_connection: AgentConnection,
         mappings: HashMap<RemotePort, LocalPort>,
         network_config: IncomingConfig,
+        idle_local_http_connection_timeout: Duration,
     ) -> Result<Self, PortForwardError> {
         let mut background_tasks: BackgroundTasks<(), ProxyMessage, IncomingProxyError> =
             Default::default();
-        let incoming = background_tasks.register(IncomingProxy::default(), (), 512);
+        let incoming = background_tasks.register(
+            IncomingProxy::new(idle_local_http_connection_timeout),
+            (),
+            512,
+        );
 
         agent_connection
             .sender
@@ -1284,12 +1289,17 @@ mod test {
         let (mut test_connection, agent_connection) = TestAgentConnection::new();
 
         tokio::spawn(async move {
-            ReversePortForwarder::new(agent_connection, mappings, network_config)
-                .await
-                .unwrap()
-                .run()
-                .await
-                .unwrap()
+            ReversePortForwarder::new(
+                agent_connection,
+                mappings,
+                network_config,
+                Duration::from_secs(3),
+            )
+            .await
+            .unwrap()
+            .run()
+            .await
+            .unwrap()
         });
 
         // expect port subscription for remote port and send subscribe result
@@ -1352,12 +1362,17 @@ mod test {
 
         let (mut test_connection, agent_connection) = TestAgentConnection::new();
         tokio::spawn(async move {
-            ReversePortForwarder::new(agent_connection, mappings, network_config)
-                .await
-                .unwrap()
-                .run()
-                .await
-                .unwrap()
+            ReversePortForwarder::new(
+                agent_connection,
+                mappings,
+                network_config,
+                Duration::from_secs(3),
+            )
+            .await
+            .unwrap()
+            .run()
+            .await
+            .unwrap()
         });
 
         // expect port subscription for remote port and send subscribe result
@@ -1436,10 +1451,14 @@ mod test {
 
         let (mut test_connection, agent_connection) = TestAgentConnection::new();
         tokio::spawn(async move {
-            let mut port_forwarder =
-                ReversePortForwarder::new(agent_connection, mappings, network_config)
-                    .await
-                    .unwrap();
+            let mut port_forwarder = ReversePortForwarder::new(
+                agent_connection,
+                mappings,
+                network_config,
+                Duration::from_secs(3),
+            )
+            .await
+            .unwrap();
             port_forwarder.run().await.unwrap()
         });
 
@@ -1548,10 +1567,14 @@ mod test {
         let (mut test_connection, agent_connection) = TestAgentConnection::new();
 
         tokio::spawn(async move {
-            let mut port_forwarder =
-                ReversePortForwarder::new(agent_connection, mappings, network_config)
-                    .await
-                    .unwrap();
+            let mut port_forwarder = ReversePortForwarder::new(
+                agent_connection,
+                mappings,
+                network_config,
+                Duration::from_secs(3),
+            )
+            .await
+            .unwrap();
             port_forwarder.run().await.unwrap()
         });
 

mirrord/config/configuration.md

@@ -496,6 +496,20 @@ but the feature is not stable and may cause other issues.
 
 Enables `getifaddrs` hook that removes IPv6 interfaces from the list returned by libc.
 
+### _experimental_ idle_local_http_connection_timeout {#experimental-idle_local_http_connection_timeout}
+
+Sets a timeout for idle local HTTP connections (in milliseconds).
+
+HTTP requests stolen with a filter are delivered to the local application
+from a HTTP connection made from the local machine. Once a request is delivered,
+the connection is cached for some time, so that it can be reused to deliver
+the next request.
+
+This timeout determines for how long such connections are cached.
+
+Set to 0 to disable caching local HTTP connections (connections will be dropped as soon as
+the request is delivered).
+
 ### _experimental_ readlink {#experimental-readlink}
 
 DEPRECATED, WILL BE REMOVED

mirrord/config/src/experimental.rs

@@ -70,6 +70,22 @@ pub struct ExperimentalConfig {
     /// <https://github.com/metalbear-co/mirrord/issues/2069>
     #[config(default = 128000)]
     pub readonly_file_buffer: u64,
+
+    /// ### _experimental_ idle_local_http_connection_timeout {#experimental-idle_local_http_connection_timeout}
+    ///
+    /// Sets a timeout for idle local HTTP connections (in milliseconds).
+    ///
+    /// HTTP requests stolen with a filter are delivered to the local application
+    /// from a HTTP connection made from the local machine. Once a request is delivered,
+    /// the connection is cached for some time, so that it can be reused to deliver
+    /// the next request.
+    ///
+    /// This timeout determines for how long such connections are cached.
+    ///
+    /// Set to 0 to disable caching local HTTP connections (connections will be dropped as soon as
+    /// the request is delivered).
+    #[config(default = 3000)]
+    pub idle_local_http_connection_timeout: u64,
 }
 
 impl CollectAnalytics for &ExperimentalConfig {
@@ -81,5 +97,9 @@ impl CollectAnalytics for &ExperimentalConfig {
         analytics.add("hide_ipv6_interfaces", self.hide_ipv6_interfaces);
         analytics.add("disable_reuseaddr", self.disable_reuseaddr);
         analytics.add("readonly_file_buffer", self.readonly_file_buffer);
+        analytics.add(
+            "idle_local_http_connection_timeout",
+            self.idle_local_http_connection_timeout,
+        );
     }
 }

mirrord/intproxy/src/lib.rs

@@ -73,6 +73,7 @@ impl IntProxy {
         agent_conn: AgentConnection,
         listener: TcpListener,
         file_buffer_size: u64,
+        idle_local_http_connection_timeout: Duration,
     ) -> Self {
         let mut background_tasks: BackgroundTasks<MainTaskId, ProxyMessage, IntProxyError> =
             Default::default();
@@ -100,7 +101,7 @@ impl IntProxy {
             Self::CHANNEL_SIZE,
         );
         let incoming = background_tasks.register(
-            IncomingProxy::default(),
+            IncomingProxy::new(idle_local_http_connection_timeout),
             MainTaskId::IncomingProxy,
             Self::CHANNEL_SIZE,
         );

mirrord/intproxy/src/proxies/incoming.rs

@@ -6,7 +6,7 @@
 //!    until connection becomes readable (is TCP) or receives an http request.
 //! 2. HttpSender -
 
-use std::{collections::HashMap, io, net::SocketAddr};
+use std::{collections::HashMap, io, net::SocketAddr, time::Duration};
 
 use bound_socket::BoundTcpSocket;
 use http::{ClientStore, ResponseMode, StreamingBody};
@@ -126,7 +126,6 @@ struct HttpGatewayHandle {
 /// An HTTP request stolen with a filter can result in an HTTP upgrade.
 /// When this happens, the TCP connection is recovered and passed to a new [`TcpProxyTask`].
 /// The TCP connection is then treated as stolen without a filter.
-#[derive(Default)]
 pub struct IncomingProxy {
     /// Active port subscriptions for all layers.
     subscriptions: SubscriptionsManager,
@@ -157,6 +156,19 @@ impl IncomingProxy {
     /// Used when registering new tasks in the internal [`BackgroundTasks`] instance.
     const CHANNEL_SIZE: usize = 512;
 
+    pub fn new(idle_local_http_connection_timeout: Duration) -> Self {
+        Self {
+            subscriptions: Default::default(),
+            metadata_store: Default::default(),
+            response_mode: Default::default(),
+            client_store: ClientStore::new_with_timeout(idle_local_http_connection_timeout),
+            mirror_tcp_proxies: Default::default(),
+            steal_tcp_proxies: Default::default(),
+            http_gateways: Default::default(),
+            tasks: Default::default(),
+        }
+    }
+
     /// Starts a new [`HttpGatewayTask`] to handle the given request.
     ///
     /// If we don't have a [`PortSubscription`] for the port, the task is not started.

mirrord/layer/tests/common/mod.rs

@@ -111,7 +111,8 @@ impl TestIntProxy {
             let agent_conn = AgentConnection::new_for_raw_address(fake_agent_address)
                 .await
                 .unwrap();
-            let intproxy = IntProxy::new_with_connection(agent_conn, listener, 0);
+            let intproxy =
+                IntProxy::new_with_connection(agent_conn, listener, 0, Duration::from_secs(3));
             intproxy
                 .run(Duration::from_secs(5), Duration::from_secs(5))
                 .await