feat: support Lambda Managed Instances

- Add Config.max_concurrency and size runtime HTTP client pool from AWS_LAMBDA_MAX_CONCURRENCY.

- Introduce windowed concurrent /next polling with semaphore-limited handler tasks and shutdown coordination.

- Require Clone + Send + 'static handlers in lambda-runtime and lambda-http, and make internal layers/HTTP adapters cloneable.

- Adjust streaming HTTP to use BoxCloneService and align bounds for concurrent execution.

- Add RIE LMI helper (Makefile + test-rie.sh) and minor robustness improvements (Context parsing, basic example error logging).

Tests: cargo +stable fmt --all; cargo +stable clippy --all-targets --all-features; cargo +stable test --all (integration test requiring TEST_ENDPOINT not configured); ./scripts/test-rie.sh basic-lambda

Author: alessandrobologna

Makefile

@@ -5,6 +5,7 @@ INTEG_API_INVOKE := RestApiUrl HttpApiUrl
 INTEG_EXTENSIONS := extension-fn extension-trait logs-trait
 # Using musl to run extensions on both AL1 and AL2
 INTEG_ARCH := x86_64-unknown-linux-musl
+RIE_MAX_CONCURRENCY ?= 4
 
 define uppercase
 $(shell sed -r 's/(^|-)(\w)/\U\2/g' <<< $(1))
@@ -111,4 +112,8 @@ fmt:
 	cargo +nightly fmt --all
 
 test-rie:
-	./scripts/test-rie.sh $(EXAMPLE)
+	./scripts/test-rie.sh $(EXAMPLE)
+
+# Run RIE in Lambda Managed Instance (LMI) mode with concurrent polling.
+test-rie-lmi:
+	RIE_MAX_CONCURRENCY=$(RIE_MAX_CONCURRENCY) ./scripts/test-rie.sh $(EXAMPLE)

examples/basic-lambda/src/main.rs

@@ -28,7 +28,10 @@ async fn main() -> Result<(), Error> {
     tracing::init_default_subscriber();
 
     let func = service_fn(my_handler);
-    lambda_runtime::run(func).await?;
+    if let Err(err) = lambda_runtime::run(func).await {
+        eprintln!("run error: {:?}", err);
+        return Err(err);
+    }
     Ok(())
 }
 

lambda-http/src/lib.rs

@@ -151,11 +151,23 @@ pub struct Adapter<'a, R, S> {
     _phantom_data: PhantomData<&'a R>,
 }
 
+impl<'a, R, S> Clone for Adapter<'a, R, S>
+where
+    S: Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            service: self.service.clone(),
+            _phantom_data: PhantomData,
+        }
+    }
+}
+
 impl<'a, R, S, E> From<S> for Adapter<'a, R, S>
 where
-    S: Service<Request, Response = R, Error = E>,
-    S::Future: Send + 'a,
-    R: IntoResponse,
+    S: Service<Request, Response = R, Error = E> + Clone + Send + 'static,
+    S::Future: Send + 'static,
+    R: IntoResponse + Send + Sync + 'static,
 {
     fn from(service: S) -> Self {
         Adapter {
@@ -193,12 +205,12 @@ where
 ///
 /// This takes care of transforming the LambdaEvent into a [`Request`] and then
 /// converting the result into a `LambdaResponse`.
-pub async fn run<'a, R, S, E>(handler: S) -> Result<(), Error>
+pub async fn run<R, S, E>(handler: S) -> Result<(), Error>
 where
-    S: Service<Request, Response = R, Error = E>,
-    S::Future: Send + 'a,
-    R: IntoResponse,
-    E: std::fmt::Debug + Into<Diagnostic>,
+    S: Service<Request, Response = R, Error = E> + Clone + Send + 'static,
+    S::Future: Send + 'static,
+    R: IntoResponse + Send + Sync + 'static,
+    E: std::fmt::Debug + Into<Diagnostic> + Send + 'static,
 {
     lambda_runtime::run(Adapter::from(handler)).await
 }

lambda-http/src/streaming.rs

@@ -9,10 +9,7 @@ use futures_util::{Stream, TryFutureExt};
 pub use http::{self, Response};
 use http_body::Body;
 use lambda_runtime::{
-    tower::{
-        util::{MapRequest, MapResponse},
-        ServiceBuilder, ServiceExt,
-    },
+    tower::{util::BoxCloneService, ServiceBuilder, ServiceExt},
     Diagnostic,
 };
 pub use lambda_runtime::{Error, LambdaEvent, MetadataPrelude, Service, StreamResponse};
@@ -27,10 +24,22 @@ pub struct StreamAdapter<'a, S, B> {
     _phantom_data: PhantomData<&'a B>,
 }
 
+impl<'a, S, B> Clone for StreamAdapter<'a, S, B>
+where
+    S: Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            service: self.service.clone(),
+            _phantom_data: PhantomData,
+        }
+    }
+}
+
 impl<'a, S, B, E> From<S> for StreamAdapter<'a, S, B>
 where
-    S: Service<Request, Response = Response<B>, Error = E>,
-    S::Future: Send + 'a,
+    S: Service<Request, Response = Response<B>, Error = E> + Clone + Send + 'static,
+    S::Future: Send + 'static,
     B: Body + Unpin + Send + 'static,
     B::Data: Into<Bytes> + Send,
     B::Error: Into<Error> + Send + Debug,
@@ -45,15 +54,15 @@ where
 
 impl<'a, S, B, E> Service<LambdaEvent<LambdaRequest>> for StreamAdapter<'a, S, B>
 where
-    S: Service<Request, Response = Response<B>, Error = E>,
-    S::Future: Send + 'a,
+    S: Service<Request, Response = Response<B>, Error = E> + Clone + Send + 'static,
+    S::Future: Send + 'static,
     B: Body + Unpin + Send + 'static,
     B::Data: Into<Bytes> + Send,
     B::Error: Into<Error> + Send + Debug,
 {
     type Response = StreamResponse<BodyStream<B>>;
     type Error = E;
-    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'a>>;
+    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.service.poll_ready(cx)
@@ -78,27 +87,23 @@ where
 /// Used internally by [`run_with_streaming_response`]; not part of the public
 /// API.
 #[allow(clippy::type_complexity)]
-fn into_stream_service<'a, S, B, E>(
+fn into_stream_service<S, B, E>(
     handler: S,
-) -> MapResponse<
-    MapRequest<S, impl FnMut(LambdaEvent<LambdaRequest>) -> Request>,
-    impl FnOnce(Response<B>) -> StreamResponse<BodyStream<B>> + Clone,
->
+) -> BoxCloneService<LambdaEvent<LambdaRequest>, StreamResponse<BodyStream<B>>, E>
 where
-    S: Service<Request, Response = Response<B>, Error = E>,
-    S::Future: Send + 'a,
-    E: Debug + Into<Diagnostic>,
+    S: Service<Request, Response = Response<B>, Error = E> + Clone + Send + 'static,
+    S::Future: Send + 'static,
+    E: Debug + Into<Diagnostic> + Send + 'static,
     B: Body + Unpin + Send + 'static,
     B::Data: Into<Bytes> + Send,
     B::Error: Into<Error> + Send + Debug,
 {
-    ServiceBuilder::new()
-        .map_request(|req: LambdaEvent<LambdaRequest>| {
-            let event: Request = req.payload.into();
-            event.with_lambda_context(req.context)
-        })
+    let svc = ServiceBuilder::new()
+        .map_request(event_to_request as fn(LambdaEvent<LambdaRequest>) -> Request)
         .service(handler)
-        .map_response(into_stream_response)
+        .map_response(into_stream_response);
+
+    BoxCloneService::new(svc)
 }
 
 /// Converts an `http::Response<B>` into a streaming Lambda response.
@@ -128,18 +133,23 @@ where
     }
 }
 
+fn event_to_request(req: LambdaEvent<LambdaRequest>) -> Request {
+    let event: Request = req.payload.into();
+    event.with_lambda_context(req.context)
+}
+
 /// Runs the Lambda runtime with a handler that returns **streaming** HTTP
 /// responses.
 ///
 /// See the [AWS docs for response streaming].
 ///
 /// [AWS docs for response streaming]:
 ///     https://docs.aws.amazon.com/lambda/latest/dg/configuration-response-streaming.html
-pub async fn run_with_streaming_response<'a, S, B, E>(handler: S) -> Result<(), Error>
+pub async fn run_with_streaming_response<S, B, E>(handler: S) -> Result<(), Error>
 where
-    S: Service<Request, Response = Response<B>, Error = E>,
-    S::Future: Send + 'a,
-    E: Debug + Into<Diagnostic>,
+    S: Service<Request, Response = Response<B>, Error = E> + Clone + Send + 'static,
+    S::Future: Send + 'static,
+    E: Debug + Into<Diagnostic> + Send + 'static,
     B: Body + Unpin + Send + 'static,
     B::Data: Into<Bytes> + Send,
     B::Error: Into<Error> + Send + Debug,

lambda-runtime-api-client/src/lib.rs

@@ -41,6 +41,7 @@ impl Client {
         ClientBuilder {
             connector: HttpConnector::new(),
             uri: None,
+            pool_size: None,
         }
     }
 }
@@ -59,11 +60,16 @@ impl Client {
         self.client.request(req).map_err(Into::into).boxed()
     }
 
-    /// Create a new client with a given base URI and HTTP connector.
-    fn with(base: Uri, connector: HttpConnector) -> Self {
-        let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new())
-            .http1_max_buf_size(1024 * 1024)
-            .build(connector);
+    /// Create a new client with a given base URI, HTTP connector, and optional pool size hint.
+    fn with(base: Uri, connector: HttpConnector, pool_size: Option<usize>) -> Self {
+        let mut builder = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new());
+        builder.http1_max_buf_size(1024 * 1024);
+
+        if let Some(size) = pool_size {
+            builder.pool_max_idle_per_host(size);
+        }
+
+        let client = builder.build(connector);
         Self { base, client }
     }
 
@@ -94,6 +100,7 @@ impl Client {
 pub struct ClientBuilder {
     connector: HttpConnector,
     uri: Option<http::Uri>,
+    pool_size: Option<usize>,
 }
 
 impl ClientBuilder {
@@ -102,13 +109,26 @@ impl ClientBuilder {
         ClientBuilder {
             connector,
             uri: self.uri,
+            pool_size: self.pool_size,
         }
     }
 
     /// Create a new builder with a given base URI.
     /// Inherits all other attributes from the existent builder.
     pub fn with_endpoint(self, uri: http::Uri) -> Self {
-        Self { uri: Some(uri), ..self }
+        Self {
+            uri: Some(uri),
+            pool_size: self.pool_size,
+            connector: self.connector,
+        }
+    }
+
+    /// Provide a pool size hint for the underlying Hyper client.
+    pub fn with_pool_size(self, pool_size: usize) -> Self {
+        Self {
+            pool_size: Some(pool_size),
+            ..self
+        }
     }
 
     /// Create the new client to interact with the Runtime API.
@@ -120,7 +140,7 @@ impl ClientBuilder {
                 uri.try_into().expect("Unable to convert to URL")
             }
         };
-        Ok(Client::with(uri, self.connector))
+        Ok(Client::with(uri, self.connector, self.pool_size))
     }
 }
 
@@ -182,4 +202,17 @@ mod tests {
             &req.uri().to_string()
         );
     }
+
+    #[test]
+    fn builder_accepts_pool_size() {
+        let base = "http://localhost:9001";
+        let expected: Uri = base.parse().unwrap();
+        let client = Client::builder()
+            .with_pool_size(4)
+            .with_endpoint(base.parse().unwrap())
+            .build()
+            .unwrap();
+
+        assert_eq!(client.base, expected);
+    }
 }

lambda-runtime/src/layers/api_client.rs

@@ -44,6 +44,18 @@ where
     }
 }
 
+impl<S> Clone for RuntimeApiClientService<S>
+where
+    S: Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+            client: self.client.clone(),
+        }
+    }
+}
+
 #[pin_project(project = RuntimeApiClientFutureProj)]
 pub enum RuntimeApiClientFuture<F> {
     First(#[pin] F, Arc<Client>),

lambda-runtime/src/layers/api_response.rs

@@ -51,6 +51,27 @@ impl<S, EventPayload, Response, BufferedResponse, StreamingResponse, StreamItem,
     }
 }
 
+impl<S, EventPayload, Response, BufferedResponse, StreamingResponse, StreamItem, StreamError> Clone
+    for RuntimeApiResponseService<
+        S,
+        EventPayload,
+        Response,
+        BufferedResponse,
+        StreamingResponse,
+        StreamItem,
+        StreamError,
+    >
+where
+    S: Clone,
+{
+    fn clone(&self) -> Self {
+        Self {
+            inner: self.inner.clone(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
 impl<S, EventPayload, Response, BufferedResponse, StreamingResponse, StreamItem, StreamError> Service<LambdaInvocation>
     for RuntimeApiResponseService<
         S,

lambda-runtime/src/layers/trace.rs

@@ -25,6 +25,7 @@ impl<S> Layer<S> for TracingLayer {
 }
 
 /// Tower service returned by [TracingLayer].
+#[derive(Clone)]
 pub struct TracingService<S> {
     inner: S,
 }