Set a limit on max number of concurrent outbound http requests

Signed-off-by: Ryan Levick <[email protected]>

Author: rylev

crates/factor-outbound-http/src/lib.rs

@@ -22,6 +22,7 @@ use spin_factors::{
     anyhow, ConfigureAppContext, Factor, FactorData, PrepareContext, RuntimeFactors,
     SelfInstanceBuilder,
 };
+use tokio::sync::Semaphore;
 use wasmtime_wasi_http::WasiHttpCtx;
 
 pub use wasmtime_wasi_http::{
@@ -51,13 +52,15 @@ impl Factor for OutboundHttpFactor {
         &self,
         mut ctx: ConfigureAppContext<T, Self>,
     ) -> anyhow::Result<Self::AppState> {
-        let connection_pooling = ctx
-            .take_runtime_config()
-            .unwrap_or_default()
-            .connection_pooling;
+        let config = ctx.take_runtime_config().unwrap_or_default();
         Ok(AppState {
-            wasi_http_clients: wasi::HttpClients::new(connection_pooling),
-            connection_pooling,
+            wasi_http_clients: wasi::HttpClients::new(config.connection_pooling_enabled),
+            connection_pooling_enabled: config.connection_pooling_enabled,
+            concurrent_outbound_requests_semaphore: config
+                .max_concurrent_requests
+                // Permit count is the max concurrent requests + 1.
+                // i.e., 0 concurrent requests means 1 total request.
+                .map(|n| Arc::new(Semaphore::new(n + 1))),
         })
     }
 
@@ -78,7 +81,11 @@ impl Factor for OutboundHttpFactor {
             request_interceptor: None,
             spin_http_client: None,
             wasi_http_clients: ctx.app_state().wasi_http_clients.clone(),
-            connection_pooling: ctx.app_state().connection_pooling,
+            connection_pooling_enabled: ctx.app_state().connection_pooling_enabled,
+            concurrent_outbound_requests_semaphore: ctx
+                .app_state()
+                .concurrent_outbound_requests_semaphore
+                .clone(),
         })
     }
 }
@@ -92,7 +99,7 @@ pub struct InstanceState {
     request_interceptor: Option<Arc<dyn OutboundHttpInterceptor>>,
     // Connection-pooling client for 'fermyon:spin/http' interface
     //
-    // TODO: We could move this to `AppState` to like the
+    // TODO: We could move this to `AppState` like the
     // `wasi:http/outgoing-handler` pool for consistency, although it's probably
     // not a high priority given that `fermyon:spin/http` is deprecated anyway.
     spin_http_client: Option<reqwest::Client>,
@@ -101,7 +108,10 @@ pub struct InstanceState {
     // This is a clone of `AppState::wasi_http_clients`, meaning it is shared
     // among all instances of the app.
     wasi_http_clients: wasi::HttpClients,
-    connection_pooling: bool,
+    /// Whether connection pooling is enabled for this instance.
+    connection_pooling_enabled: bool,
+    /// A semaphore to limit the number of concurrent outbound requests.
+    concurrent_outbound_requests_semaphore: Option<Arc<Semaphore>>,
 }
 
 impl InstanceState {
@@ -183,5 +193,8 @@ impl std::fmt::Display for SelfRequestOrigin {
 pub struct AppState {
     // Connection pooling clients for `wasi:http/outgoing-handler` interface
     wasi_http_clients: wasi::HttpClients,
-    connection_pooling: bool,
+    /// Whether connection pooling is enabled for this app.
+    connection_pooling_enabled: bool,
+    /// A semaphore to limit the number of concurrent outbound requests.
+    concurrent_outbound_requests_semaphore: Option<Arc<Semaphore>>,
 }

crates/factor-outbound-http/src/runtime_config.rs

@@ -5,13 +5,16 @@ pub mod spin;
 #[derive(Debug)]
 pub struct RuntimeConfig {
     /// If true, enable connection pooling and reuse.
-    pub connection_pooling: bool,
+    pub connection_pooling_enabled: bool,
+    /// If set, limits the number of concurrent outbound requests.
+    pub max_concurrent_requests: Option<usize>,
 }
 
 impl Default for RuntimeConfig {
     fn default() -> Self {
         Self {
-            connection_pooling: true,
+            connection_pooling_enabled: true,
+            max_concurrent_requests: None,
         }
     }
 }

crates/factor-outbound-http/src/runtime_config/spin.rs

@@ -6,17 +6,17 @@ use spin_factors::runtime_config::toml::GetTomlValue;
 /// Expects table to be in the format:
 /// ```toml
 /// [outbound_http]
-/// connection_pooling = true
+/// connection_pooling = true # optional, defaults to true
+/// max_concurrent_requests = 10 # optional, defaults to unlimited
 /// ```
 pub fn config_from_table(
     table: &impl GetTomlValue,
 ) -> anyhow::Result<Option<super::RuntimeConfig>> {
     if let Some(outbound_http) = table.get("outbound_http") {
+        let outbound_http_toml = outbound_http.clone().try_into::<OutboundHttpToml>()?;
         Ok(Some(super::RuntimeConfig {
-            connection_pooling: outbound_http
-                .clone()
-                .try_into::<OutboundHttpToml>()?
-                .connection_pooling,
+            connection_pooling_enabled: outbound_http_toml.connection_pooling,
+            max_concurrent_requests: outbound_http_toml.max_concurrent_requests,
         }))
     } else {
         Ok(None)
@@ -28,4 +28,6 @@ pub fn config_from_table(
 struct OutboundHttpToml {
     #[serde(default)]
     connection_pooling: bool,
+    #[serde(default)]
+    max_concurrent_requests: Option<usize>,
 }

crates/factor-outbound-http/src/spin.rs

@@ -21,7 +21,6 @@ impl spin_http::Host for crate::InstanceState {
         if !req.params.is_empty() {
             tracing::warn!("HTTP params field is deprecated");
         }
-
         let req_url = if !uri.starts_with('/') {
             // Absolute URI
             let is_allowed = self
@@ -92,13 +91,19 @@ impl spin_http::Host for crate::InstanceState {
         // in a single component execution
         let client = self.spin_http_client.get_or_insert_with(|| {
             let mut builder = reqwest::Client::builder();
-            if !self.connection_pooling {
+            if !self.connection_pooling_enabled {
                 builder = builder.pool_max_idle_per_host(0);
             }
             builder.build().unwrap()
         });
 
+        // If we're limiting concurrent outbound requests, acquire a permit
+        let permit = match &self.concurrent_outbound_requests_semaphore {
+            Some(s) => s.acquire().await.ok(),
+            None => None,
+        };
         let resp = client.execute(req).await.map_err(log_reqwest_error)?;
+        drop(permit);
 
         tracing::trace!("Returning response from outbound request to {req_url}");
         span.record("http.response.status_code", resp.status().as_u16());

crates/factor-outbound-http/src/wasi.rs

@@ -26,6 +26,7 @@ use spin_factors::{wasmtime::component::ResourceTable, RuntimeFactorsInstanceSta
 use tokio::{
     io::{AsyncRead, AsyncWrite, ReadBuf},
     net::TcpStream,
+    sync::Semaphore,
     time::timeout,
 };
 use tokio_rustls::client::TlsStream;
@@ -128,6 +129,10 @@ impl WasiHttpView for WasiHttpImplInner<'_> {
             self_request_origin: self.state.self_request_origin.clone(),
             blocked_networks: self.state.blocked_networks.clone(),
             http_clients: self.state.wasi_http_clients.clone(),
+            concurrent_outbound_requests_semaphore: self
+                .state
+                .concurrent_outbound_requests_semaphore
+                .clone(),
         };
         Ok(HostFutureIncomingResponse::Pending(
             wasmtime_wasi::runtime::spawn(
@@ -153,6 +158,7 @@ struct RequestSender {
     self_request_origin: Option<SelfRequestOrigin>,
     request_interceptor: Option<Arc<dyn OutboundHttpInterceptor>>,
     http_clients: HttpClients,
+    concurrent_outbound_requests_semaphore: Option<Arc<Semaphore>>,
 }
 
 impl RequestSender {
@@ -320,11 +326,17 @@ impl RequestSender {
             },
         );
 
+        // If we're limiting concurrent outbound requests, acquire a permit
+        let permit = match &self.concurrent_outbound_requests_semaphore {
+            Some(s) => s.acquire().await.ok(),
+            None => None,
+        };
         let resp = timeout(first_byte_timeout, resp)
             .await
             .map_err(|_| ErrorCode::ConnectionReadTimeout)?
             .map_err(hyper_legacy_request_error)?
             .map(|body| body.map_err(hyper_request_error).boxed());
+        drop(permit);
 
         tracing::Span::current().record("http.response.status_code", resp.status().as_u16());