refactor(lambda-managed-instances): warn on run() with `AWS_LAMBDA_MAX_CONCURRENCY, rename feature experimental-concurrency -> concurrency->tokio

Author: jlizen

examples/basic-lambda-concurrent/Cargo.toml

@@ -4,6 +4,6 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-lambda_runtime = { path = "../../lambda-runtime", features = ["experimental-concurrency"] }
+lambda_runtime = { path = "../../lambda-runtime", features = ["concurrency-tokio"] }
 serde = "1.0.219"
 tokio = { version = "1", features = ["macros"] }

lambda-http/Cargo.toml

@@ -29,7 +29,7 @@ opentelemetry = ["lambda_runtime/opentelemetry"] # enables access to the OpenTel
 anyhow = ["lambda_runtime/anyhow"] # enables From<T> for Diagnostic for anyhow error types, see README.md for more info
 eyre = ["lambda_runtime/eyre"] # enables From<T> for Diagnostic for eyre error types, see README.md for more info
 miette = ["lambda_runtime/miette"] # enables From<T> for Diagnostic for miette error types, see README.md for more info
-experimental-concurrency = ["lambda_runtime/experimental-concurrency"]
+concurrency-tokio = ["lambda_runtime/concurrency-tokio"]
 
 [dependencies]
 bytes = { workspace = true }

lambda-http/src/lib.rs

@@ -102,7 +102,7 @@ use std::{
 };
 
 mod streaming;
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 pub use streaming::run_with_streaming_response_concurrent;
 pub use streaming::{run_with_streaming_response, StreamAdapter};
 
@@ -213,7 +213,7 @@ where
 /// # Managed concurrency
 /// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, this function returns an error because
 /// it does not enable concurrent polling. If your handler can satisfy `Clone`,
-/// prefer [`run_concurrent`] (requires the `experimental-concurrency` feature),
+/// prefer [`run_concurrent`] (requires the `concurrency-tokio` feature),
 /// which honors managed concurrency and falls back to sequential behavior when
 /// unset.
 pub async fn run<'a, R, S, E>(handler: S) -> Result<(), Error>
@@ -226,18 +226,32 @@ where
     lambda_runtime::run(Adapter::from(handler)).await
 }
 
-/// Starts the Lambda Rust runtime in a mode that is compatible with
-/// Lambda Managed Instances (concurrent invocations).
+/// Starts the Lambda Rust runtime and begins polling for events on the [Lambda
+/// Runtime APIs](https://docs.aws.amazon.com/lambda/latest/dg/runtimes-api.html).
 ///
-/// Requires the `experimental-concurrency` feature.
+/// This takes care of transforming the LambdaEvent into a [`Request`] and then
+/// converting the result into a `LambdaResponse`.
+///
+/// # Managed concurrency
 ///
 /// When `AWS_LAMBDA_MAX_CONCURRENCY` is set to a value greater than 1, this
-/// will spawn `AWS_LAMBDA_MAX_CONCURRENCY` worker tasks, each running its own
-/// `/next` polling loop. When the environment variable is unset or `<= 1`,
-/// it falls back to the same sequential behavior as [`run`], so the same
-/// handler can run on both classic Lambda and Lambda Managed Instances.
-#[cfg(feature = "experimental-concurrency")]
-#[cfg_attr(docsrs, doc(cfg(feature = "experimental-concurrency")))]
+/// function spawns multiple worker tasks to handle concurrent invocations.
+/// When the environment variable is unset or `<= 1`, it falls back to
+/// sequential behavior, so the same handler can run on both classic Lambda
+/// and Lambda Managed Instances.
+///
+/// This feature requires `Clone + Send` bounds on the handler. If your handler
+/// cannot satisfy these bounds, use `run()` for sequential processing.
+///
+/// # Panics
+///
+/// This function panics if:
+/// - Called outside of a Tokio runtime with `AWS_LAMBDA_MAX_CONCURRENCY > 1`
+/// - Required Lambda environment variables are missing (`AWS_LAMBDA_FUNCTION_NAME`,
+///   `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`, `AWS_LAMBDA_FUNCTION_VERSION`,
+///   `AWS_LAMBDA_RUNTIME_API`)
+#[cfg(feature = "concurrency-tokio")]
+#[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
 pub async fn run_concurrent<R, S, E>(handler: S) -> Result<(), Error>
 where
     S: Service<Request, Response = R, Error = E> + Clone + Send + 'static,

lambda-http/src/streaming.rs

@@ -114,10 +114,10 @@ where
 
 /// Builds a streaming-aware Tower service from a `Service<Request>` that can be
 /// cloned and sent across tasks. This is used by the concurrent HTTP entrypoint.
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 type EventToRequest = fn(LambdaEvent<LambdaRequest>) -> Request;
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 #[allow(clippy::type_complexity)]
 fn into_stream_service_cloneable<S, B, E>(
     handler: S,
@@ -178,7 +178,7 @@ fn event_to_request(req: LambdaEvent<LambdaRequest>) -> Request {
 /// # Managed concurrency
 /// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, this function returns an error because
 /// it does not enable concurrent polling. Use [`run_with_streaming_response_concurrent`]
-/// (requires the `experimental-concurrency` feature) instead.
+/// (requires the `concurrency-tokio` feature) instead.
 ///
 /// [AWS docs for response streaming]:
 ///     https://docs.aws.amazon.com/lambda/latest/dg/configuration-response-streaming.html
@@ -197,13 +197,13 @@ where
 /// Runs the Lambda runtime with a handler that returns **streaming** HTTP
 /// responses, in a mode that is compatible with Lambda Managed Instances.
 ///
-/// Requires the `experimental-concurrency` feature.
+/// Requires the `concurrency-tokio` feature.
 ///
 /// This uses a cloneable, boxed service internally so it can be driven by the
 /// concurrent runtime. When `AWS_LAMBDA_MAX_CONCURRENCY` is not set or `<= 1`,
 /// it falls back to the same sequential behavior as [`run_with_streaming_response`].
-#[cfg(feature = "experimental-concurrency")]
-#[cfg_attr(docsrs, doc(cfg(feature = "experimental-concurrency")))]
+#[cfg(feature = "concurrency-tokio")]
+#[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
 pub async fn run_with_streaming_response_concurrent<S, B, E>(handler: S) -> Result<(), Error>
 where
     S: Service<Request, Response = Response<B>, Error = E> + Clone + Send + 'static,

lambda-runtime/Cargo.toml

@@ -25,7 +25,7 @@ miette = ["dep:miette"] # enables From<T> for Diagnostic for miette error types,
 # as well as default features
 # https://github.com/aws/aws-lambda-rust-runtime/issues/984
 graceful-shutdown = ["tokio/rt", "tokio/signal", "dep:lambda-extension"]
-experimental-concurrency = []
+concurrency-tokio = []
 
 [dependencies]
 anyhow = { version = "1.0.86", optional = true }

lambda-runtime/src/lib.rs

@@ -97,7 +97,7 @@ where
 /// # Managed concurrency
 /// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, this function returns an error because
 /// it does not enable concurrent polling. If your handler can satisfy `Clone`,
-/// prefer [`run_concurrent`] (requires the `experimental-concurrency` feature),
+/// prefer [`run_concurrent`] (requires the `concurrency-tokio` feature),
 /// which honors managed concurrency and falls back to sequential behavior when
 /// unset.
 ///
@@ -136,7 +136,7 @@ where
 /// Starts the Lambda Rust runtime in a mode that is compatible with
 /// Lambda Managed Instances (concurrent invocations).
 ///
-/// Requires the `experimental-concurrency` feature.
+/// Requires the `concurrency-tokio` feature.
 ///
 /// When `AWS_LAMBDA_MAX_CONCURRENCY` is set to a value greater than 1, this
 /// will spawn `AWS_LAMBDA_MAX_CONCURRENCY` worker tasks, each running its own
@@ -163,8 +163,8 @@ where
 ///     Ok(event.payload)
 /// }
 /// ```
-#[cfg(feature = "experimental-concurrency")]
-#[cfg_attr(docsrs, doc(cfg(feature = "experimental-concurrency")))]
+#[cfg(feature = "concurrency-tokio")]
+#[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
 pub async fn run_concurrent<A, F, R, B, S, D, E>(handler: F) -> Result<(), Error>
 where
     F: Service<LambdaEvent<A>, Response = R> + Clone + Send + 'static,

lambda-runtime/src/runtime.rs

@@ -4,18 +4,18 @@ use crate::{
     types::{invoke_request_id, IntoFunctionResponse, LambdaEvent},
     Config, Context, Diagnostic,
 };
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 use futures::stream::FuturesUnordered;
 use http_body_util::BodyExt;
 use lambda_runtime_api_client::{BoxError, Client as ApiClient};
 use serde::{Deserialize, Serialize};
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 use std::fmt;
 use std::{env, fmt::Debug, future::Future, io, sync::Arc};
 use tokio_stream::{Stream, StreamExt};
 use tower::{Layer, Service, ServiceExt};
 use tracing::trace;
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 use tracing::{debug, error, info_span, warn, Instrument};
 
 /* ----------------------------------------- INVOCATION ---------------------------------------- */
@@ -96,6 +96,13 @@ where
     /// Note that manually creating a [Runtime] does not add tracing to the executed handler
     /// as is done by [super::run]. If you want to add the default tracing functionality, call
     /// [Runtime::layer] with a [super::layers::TracingLayer].
+    ///
+    ///
+    /// # Panics
+    ///
+    /// This function panics if required Lambda environment variables are missing
+    /// (`AWS_LAMBDA_FUNCTION_NAME`, `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`,
+    /// `AWS_LAMBDA_FUNCTION_VERSION`, `AWS_LAMBDA_RUNTIME_API`).
     pub fn new(handler: F) -> Self {
         trace!("Loading config from env");
         let config = Arc::new(Config::from_env());
@@ -154,18 +161,28 @@ impl<S> Runtime<S> {
     }
 }
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 impl<S> Runtime<S>
 where
     S: Service<LambdaInvocation, Response = (), Error = BoxError> + Clone + Send + 'static,
     S::Future: Send,
 {
-    /// Start the runtime in concurrent mode when configured for Lambda managed-concurrency.
+    /// Start the runtime and begin polling for events on the Lambda Runtime API, in a mode
+    /// that supports Lambda-managed concurrency.
+    ///
+    /// When `AWS_LAMBDA_MAX_CONCURRENCY` is set to a value greater than 1, this
+    /// spawns multiple worker tasks to handle concurrent invocations. When the
+    /// environment variable is unset or `<= 1`, it falls back to sequential
+    /// behavior, so the same handler can run on both classic Lambda and Lambda
+    /// Managed Instances.
     ///
-    /// If `AWS_LAMBDA_MAX_CONCURRENCY` is not set or is `<= 1`, this falls back to the
-    /// sequential `run_with_incoming` loop so that the same handler can run on both
-    /// classic Lambda and Lambda Managed Instances.
-    #[cfg_attr(docsrs, doc(cfg(feature = "experimental-concurrency")))]
+    /// This feature requires `Clone + Send + 'static` bounds on the service. If your handler
+    /// cannot satisfy these bounds, use [Runtime::run] instead.
+    ///
+    /// # Panics
+    ///
+    /// This function panics if called outside of a Tokio runtime with `AWS_LAMBDA_MAX_CONCURRENCY > 1`
+    #[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
     pub async fn run_concurrent(self) -> Result<(), BoxError> {
         if self.concurrency_limit > 1 {
             trace!("Concurrent mode: _X_AMZN_TRACE_ID is not set; use context.xray_trace_id");
@@ -259,20 +276,20 @@ where
     }
 }
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 #[derive(Debug)]
 enum WorkerError {
     CleanExit(tokio::task::Id),
     Failure(tokio::task::Id, BoxError),
 }
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 #[derive(Debug)]
 struct ConcurrentWorkerErrors {
     errors: Vec<WorkerError>,
 }
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 #[derive(Serialize)]
 struct ConcurrentWorkerErrorsPayload<'a> {
     message: &'a str,
@@ -282,14 +299,14 @@ struct ConcurrentWorkerErrorsPayload<'a> {
     failures: Vec<WorkerFailurePayload>,
 }
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 #[derive(Serialize)]
 struct WorkerFailurePayload {
     id: String,
     err: String,
 }
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 impl fmt::Display for ConcurrentWorkerErrors {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let mut clean = Vec::new();
@@ -326,7 +343,7 @@ impl fmt::Display for ConcurrentWorkerErrors {
     }
 }
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 impl std::error::Error for ConcurrentWorkerErrors {}
 
 impl<S> Runtime<S>
@@ -335,14 +352,17 @@ where
 {
     /// Start the runtime and begin polling for events on the Lambda Runtime API.
     ///
-    /// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, this returns an error because it does not enable
-    /// concurrent polling. Enable the `experimental-concurrency` feature and use
-    /// [`Runtime::run_concurrent`] instead.
+    /// The runtime will process requests sequentially.
+    ///
+    /// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, a warning is logged.
+    /// Enable the `concurrency-tokio` feature and use [`Runtime::run_concurrent`]
+    /// to enable concurrent request processing with Lambda Managed Instances.
     pub async fn run(self) -> Result<(), BoxError> {
         if let Some(raw) = concurrency_env_value() {
-            return Err(Box::new(io::Error::other(format!(
-                "AWS_LAMBDA_MAX_CONCURRENCY is set to '{raw}', but Runtime::run does not support concurrent polling; enable the experimental-concurrency feature and use Runtime::run_concurrent instead"
-            ))));
+            tracing::warn!(
+                "AWS_LAMBDA_MAX_CONCURRENCY is set to '{}', but the concurrency-tokio feature is not enabled; running sequentially",
+                raw
+            );
         }
         let incoming = incoming(&self.client);
         Self::run_with_incoming(self.service, self.config, incoming).await
@@ -412,7 +432,7 @@ fn incoming(
 }
 
 /// Creates a future that polls the `/next` endpoint.
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 async fn next_event_future(client: &ApiClient) -> Result<http::Response<hyper::body::Incoming>, BoxError> {
     let req = NextEventRequest.into_req()?;
     client.call(req).await
@@ -429,7 +449,7 @@ fn concurrency_env_value() -> Option<String> {
     env::var("AWS_LAMBDA_MAX_CONCURRENCY").ok()
 }
 
-#[cfg(feature = "experimental-concurrency")]
+#[cfg(feature = "concurrency-tokio")]
 async fn concurrent_worker_loop<S>(mut service: S, config: Arc<Config>, client: Arc<ApiClient>) -> Result<(), BoxError>
 where
     S: Service<LambdaInvocation, Response = (), Error = BoxError>,
@@ -760,7 +780,7 @@ mod endpoint_tests {
         .await
     }
 
-    #[cfg(feature = "experimental-concurrency")]
+    #[cfg(feature = "concurrency-tokio")]
     #[tokio::test]
     async fn concurrent_worker_crash_does_not_stop_other_workers() -> Result<(), Error> {
         let next_calls = Arc::new(AtomicUsize::new(0));
@@ -910,7 +930,7 @@ mod endpoint_tests {
     }
 
     #[tokio::test]
-    #[cfg(feature = "experimental-concurrency")]
+    #[cfg(feature = "concurrency-tokio")]
     async fn test_concurrent_structured_logging_isolation() -> Result<(), Error> {
         use std::collections::HashSet;
         use tracing::info;