Skip to content

Commit e1a4163

Browse files
jlizenjlizen
committed
refactor(lambda-managed-instances): warn on run() with `AWS_LAMBDA_MAX_CONCURRENCY, rename feature experimental-concurrency -> concurrency->tokio
1 parent 079087d commit e1a4163

7 files changed

Lines changed: 126 additions & 58 deletions

File tree

‎examples/basic-lambda-concurrent/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,6 @@ version = "0.1.0"
44
edition = "2021"
55

66
[dependencies]
7-
lambda_runtime = { path = "../../lambda-runtime", features = ["experimental-concurrency"] }
7+
lambda_runtime = { path = "../../lambda-runtime", features = ["concurrency-tokio"] }
88
serde = "1.0.219"
99
tokio = { version = "1", features = ["macros"] }

‎lambda-http/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ opentelemetry = ["lambda_runtime/opentelemetry"] # enables access to the OpenTel
2929
anyhow = ["lambda_runtime/anyhow"] # enables From<T> for Diagnostic for anyhow error types, see README.md for more info
3030
eyre = ["lambda_runtime/eyre"] # enables From<T> for Diagnostic for eyre error types, see README.md for more info
3131
miette = ["lambda_runtime/miette"] # enables From<T> for Diagnostic for miette error types, see README.md for more info
32-
experimental-concurrency = ["lambda_runtime/experimental-concurrency"]
32+
concurrency-tokio = ["lambda_runtime/concurrency-tokio"]
3333

3434
[dependencies]
3535
bytes = { workspace = true }

‎lambda-http/src/lib.rs‎

Lines changed: 30 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ use std::{
102102
};
103103

104104
mod streaming;
105-
#[cfg(feature = "experimental-concurrency")]
105+
#[cfg(feature = "concurrency-tokio")]
106106
pub use streaming::run_with_streaming_response_concurrent;
107107
pub use streaming::{run_with_streaming_response, StreamAdapter};
108108

@@ -211,11 +211,17 @@ where
211211
/// converting the result into a `LambdaResponse`.
212212
///
213213
/// # Managed concurrency
214-
/// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, this function returns an error because
215-
/// it does not enable concurrent polling. If your handler can satisfy `Clone`,
216-
/// prefer [`run_concurrent`] (requires the `experimental-concurrency` feature),
214+
/// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, a warning is logged.
215+
/// If your handler can satisfy `Clone + Send + 'static`,
216+
/// prefer [`run_concurrent`] (requires the `concurrency-tokio` feature),
217217
/// which honors managed concurrency and falls back to sequential behavior when
218218
/// unset.
219+
///
220+
/// # Panics
221+
///
222+
/// This function panics if required Lambda environment variables are missing
223+
/// (`AWS_LAMBDA_FUNCTION_NAME`, `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`,
224+
/// `AWS_LAMBDA_FUNCTION_VERSION`, `AWS_LAMBDA_RUNTIME_API`).
219225
pub async fn run<'a, R, S, E>(handler: S) -> Result<(), Error>
220226
where
221227
S: Service<Request, Response = R, Error = E>,
@@ -226,18 +232,29 @@ where
226232
lambda_runtime::run(Adapter::from(handler)).await
227233
}
228234

229-
/// Starts the Lambda Rust runtime in a mode that is compatible with
230-
/// Lambda Managed Instances (concurrent invocations).
235+
/// Starts the Lambda Rust runtime and begins polling for events on the [Lambda
236+
/// Runtime APIs](https://docs.aws.amazon.com/lambda/latest/dg/runtimes-api.html).
237+
///
238+
/// This takes care of transforming the LambdaEvent into a [`Request`] and then
239+
/// converting the result into a `LambdaResponse`.
231240
///
232-
/// Requires the `experimental-concurrency` feature.
241+
/// # Managed concurrency
233242
///
234243
/// When `AWS_LAMBDA_MAX_CONCURRENCY` is set to a value greater than 1, this
235-
/// will spawn `AWS_LAMBDA_MAX_CONCURRENCY` worker tasks, each running its own
236-
/// `/next` polling loop. When the environment variable is unset or `<= 1`,
237-
/// it falls back to the same sequential behavior as [`run`], so the same
238-
/// handler can run on both classic Lambda and Lambda Managed Instances.
239-
#[cfg(feature = "experimental-concurrency")]
240-
#[cfg_attr(docsrs, doc(cfg(feature = "experimental-concurrency")))]
244+
/// function spawns multiple tokio worker tasks to handle concurrent invocations.
245+
/// When the environment variable is unset or `<= 1`, it falls back to
246+
/// sequential behavior, so the same handler can run on both classic Lambda
247+
/// and Lambda Managed Instances.
248+
///
249+
/// # Panics
250+
///
251+
/// This function panics if:
252+
/// - Called outside of a Tokio runtime with `AWS_LAMBDA_MAX_CONCURRENCY > 1`
253+
/// - Required Lambda environment variables are missing (`AWS_LAMBDA_FUNCTION_NAME`,
254+
/// `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`, `AWS_LAMBDA_FUNCTION_VERSION`,
255+
/// `AWS_LAMBDA_RUNTIME_API`)
256+
#[cfg(feature = "concurrency-tokio")]
257+
#[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
241258
pub async fn run_concurrent<R, S, E>(handler: S) -> Result<(), Error>
242259
where
243260
S: Service<Request, Response = R, Error = E> + Clone + Send + 'static,

‎lambda-http/src/streaming.rs‎

Lines changed: 22 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -114,10 +114,10 @@ where
114114

115115
/// Builds a streaming-aware Tower service from a `Service<Request>` that can be
116116
/// cloned and sent across tasks. This is used by the concurrent HTTP entrypoint.
117-
#[cfg(feature = "experimental-concurrency")]
117+
#[cfg(feature = "concurrency-tokio")]
118118
type EventToRequest = fn(LambdaEvent<LambdaRequest>) -> Request;
119119

120-
#[cfg(feature = "experimental-concurrency")]
120+
#[cfg(feature = "concurrency-tokio")]
121121
#[allow(clippy::type_complexity)]
122122
fn into_stream_service_cloneable<S, B, E>(
123123
handler: S,
@@ -178,10 +178,16 @@ fn event_to_request(req: LambdaEvent<LambdaRequest>) -> Request {
178178
/// # Managed concurrency
179179
/// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, this function returns an error because
180180
/// it does not enable concurrent polling. Use [`run_with_streaming_response_concurrent`]
181-
/// (requires the `experimental-concurrency` feature) instead.
181+
/// (requires the `concurrency-tokio` feature) instead.
182182
///
183183
/// [AWS docs for response streaming]:
184184
/// https://docs.aws.amazon.com/lambda/latest/dg/configuration-response-streaming.html
185+
///
186+
/// # Panics
187+
///
188+
/// This function panics if required Lambda environment variables are missing
189+
/// (`AWS_LAMBDA_FUNCTION_NAME`, `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`,
190+
/// `AWS_LAMBDA_FUNCTION_VERSION`, `AWS_LAMBDA_RUNTIME_API`).
185191
pub async fn run_with_streaming_response<'a, S, B, E>(handler: S) -> Result<(), Error>
186192
where
187193
S: Service<Request, Response = Response<B>, Error = E>,
@@ -197,13 +203,20 @@ where
197203
/// Runs the Lambda runtime with a handler that returns **streaming** HTTP
198204
/// responses, in a mode that is compatible with Lambda Managed Instances.
199205
///
200-
/// Requires the `experimental-concurrency` feature.
206+
/// When `AWS_LAMBDA_MAX_CONCURRENCY` is set to a value greater than 1, this
207+
/// spawns multiple tokio worker tasks to handle concurrent invocations. When the
208+
/// environment variable is unset or `<= 1`, it falls back to sequential
209+
/// behavior, so the same handler can run on both classic Lambda and Lambda
210+
/// Managed Instances.
211+
///
212+
/// # Panics
201213
///
202-
/// This uses a cloneable, boxed service internally so it can be driven by the
203-
/// concurrent runtime. When `AWS_LAMBDA_MAX_CONCURRENCY` is not set or `<= 1`,
204-
/// it falls back to the same sequential behavior as [`run_with_streaming_response`].
205-
#[cfg(feature = "experimental-concurrency")]
206-
#[cfg_attr(docsrs, doc(cfg(feature = "experimental-concurrency")))]
214+
/// This function panics if:
215+
/// - Called outside of a Tokio runtime
216+
/// - Required Lambda environment variables are missing (`AWS_LAMBDA_FUNCTION_NAME`,
217+
/// `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`, `AWS_LAMBDA_FUNCTION_VERSION`,
218+
/// `AWS_LAMBDA_RUNTIME_API`)
219+
#[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
207220
pub async fn run_with_streaming_response_concurrent<S, B, E>(handler: S) -> Result<(), Error>
208221
where
209222
S: Service<Request, Response = Response<B>, Error = E> + Clone + Send + 'static,

‎lambda-runtime/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@ miette = ["dep:miette"] # enables From<T> for Diagnostic for miette error types,
2525
# as well as default features
2626
# https://github.com/aws/aws-lambda-rust-runtime/issues/984
2727
graceful-shutdown = ["tokio/rt", "tokio/signal", "dep:lambda-extension"]
28-
experimental-concurrency = []
28+
concurrency-tokio = []
2929

3030
[dependencies]
3131
anyhow = { version = "1.0.86", optional = true }

‎lambda-runtime/src/lib.rs‎

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -95,9 +95,9 @@ where
9595
/// [Runtime] type directly.
9696
///
9797
/// # Managed concurrency
98-
/// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, this function returns an error because
99-
/// it does not enable concurrent polling. If your handler can satisfy `Clone`,
100-
/// prefer [`run_concurrent`] (requires the `experimental-concurrency` feature),
98+
/// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, a warning is logged.
99+
/// If your handler can satisfy `Clone + Send + 'static`,
100+
/// prefer [`run_concurrent`] (requires the `concurrency-tokio` feature),
101101
/// which honors managed concurrency and falls back to sequential behavior when
102102
/// unset.
103103
///
@@ -117,6 +117,12 @@ where
117117
/// Ok(event.payload)
118118
/// }
119119
/// ```
120+
///
121+
/// # Panics
122+
///
123+
/// This function panics if required Lambda environment variables are missing
124+
/// (`AWS_LAMBDA_FUNCTION_NAME`, `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`,
125+
/// `AWS_LAMBDA_FUNCTION_VERSION`, `AWS_LAMBDA_RUNTIME_API`).
120126
pub async fn run<A, F, R, B, S, D, E>(handler: F) -> Result<(), Error>
121127
where
122128
F: Service<LambdaEvent<A>, Response = R>,
@@ -136,10 +142,10 @@ where
136142
/// Starts the Lambda Rust runtime in a mode that is compatible with
137143
/// Lambda Managed Instances (concurrent invocations).
138144
///
139-
/// Requires the `experimental-concurrency` feature.
145+
/// Requires the `concurrency-tokio` feature.
140146
///
141147
/// When `AWS_LAMBDA_MAX_CONCURRENCY` is set to a value greater than 1, this
142-
/// will spawn `AWS_LAMBDA_MAX_CONCURRENCY` worker tasks, each running its own
148+
/// will spawn `AWS_LAMBDA_MAX_CONCURRENCY` tokio worker tasks, each running its own
143149
/// `/next` polling loop. When the environment variable is unset or `<= 1`, it
144150
/// falls back to the same sequential behavior as [`run`], so the same handler
145151
/// can run on both classic Lambda and Lambda Managed Instances.
@@ -163,8 +169,16 @@ where
163169
/// Ok(event.payload)
164170
/// }
165171
/// ```
166-
#[cfg(feature = "experimental-concurrency")]
167-
#[cfg_attr(docsrs, doc(cfg(feature = "experimental-concurrency")))]
172+
///
173+
/// # Panics
174+
///
175+
/// This function panics if:
176+
/// - Called outside of a Tokio runtime
177+
/// - Required Lambda environment variables are missing (`AWS_LAMBDA_FUNCTION_NAME`,
178+
/// `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`, `AWS_LAMBDA_FUNCTION_VERSION`,
179+
/// `AWS_LAMBDA_RUNTIME_API`)
180+
#[cfg(feature = "concurrency-tokio")]
181+
#[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
168182
pub async fn run_concurrent<A, F, R, B, S, D, E>(handler: F) -> Result<(), Error>
169183
where
170184
F: Service<LambdaEvent<A>, Response = R> + Clone + Send + 'static,

‎lambda-runtime/src/runtime.rs‎

Lines changed: 50 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -4,18 +4,18 @@ use crate::{
44
types::{invoke_request_id, IntoFunctionResponse, LambdaEvent},
55
Config, Context, Diagnostic,
66
};
7-
#[cfg(feature = "experimental-concurrency")]
7+
#[cfg(feature = "concurrency-tokio")]
88
use futures::stream::FuturesUnordered;
99
use http_body_util::BodyExt;
1010
use lambda_runtime_api_client::{BoxError, Client as ApiClient};
1111
use serde::{Deserialize, Serialize};
12-
#[cfg(feature = "experimental-concurrency")]
12+
#[cfg(feature = "concurrency-tokio")]
1313
use std::fmt;
14-
use std::{env, fmt::Debug, future::Future, io, sync::Arc};
14+
use std::{env, fmt::Debug, future::Future, sync::Arc};
1515
use tokio_stream::{Stream, StreamExt};
1616
use tower::{Layer, Service, ServiceExt};
1717
use tracing::trace;
18-
#[cfg(feature = "experimental-concurrency")]
18+
#[cfg(feature = "concurrency-tokio")]
1919
use tracing::{debug, error, info_span, warn, Instrument};
2020

2121
/* ----------------------------------------- INVOCATION ---------------------------------------- */
@@ -96,6 +96,13 @@ where
9696
/// Note that manually creating a [Runtime] does not add tracing to the executed handler
9797
/// as is done by [super::run]. If you want to add the default tracing functionality, call
9898
/// [Runtime::layer] with a [super::layers::TracingLayer].
99+
///
100+
///
101+
/// # Panics
102+
///
103+
/// This function panics if required Lambda environment variables are missing
104+
/// (`AWS_LAMBDA_FUNCTION_NAME`, `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`,
105+
/// `AWS_LAMBDA_FUNCTION_VERSION`, `AWS_LAMBDA_RUNTIME_API`).
99106
pub fn new(handler: F) -> Self {
100107
trace!("Loading config from env");
101108
let config = Arc::new(Config::from_env());
@@ -154,19 +161,30 @@ impl<S> Runtime<S> {
154161
}
155162
}
156163

157-
#[cfg(feature = "experimental-concurrency")]
164+
#[cfg(feature = "concurrency-tokio")]
158165
impl<S> Runtime<S>
159166
where
160167
S: Service<LambdaInvocation, Response = (), Error = BoxError> + Clone + Send + 'static,
161168
S::Future: Send,
162169
{
163-
/// Start the runtime in concurrent mode when configured for Lambda managed-concurrency.
170+
/// Start the runtime and begin polling for events on the Lambda Runtime API,
171+
/// in a mode that is compatible with Lambda Managed Instances.
172+
///
173+
/// When `AWS_LAMBDA_MAX_CONCURRENCY` is set to a value greater than 1, this
174+
/// spawns multiple tokio worker tasks to handle concurrent invocations. When the
175+
/// environment variable is unset or `<= 1`, it falls back to sequential
176+
/// behavior, so the same handler can run on both classic Lambda and Lambda
177+
/// Managed Instances.
178+
///
179+
/// # Panics
164180
///
165-
/// If `AWS_LAMBDA_MAX_CONCURRENCY` is not set or is `<= 1`, this falls back to the
166-
/// sequential `run_with_incoming` loop so that the same handler can run on both
167-
/// classic Lambda and Lambda Managed Instances.
168-
#[cfg_attr(docsrs, doc(cfg(feature = "experimental-concurrency")))]
181+
/// This function panics if called outside of a Tokio runtime.
182+
#[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
169183
pub async fn run_concurrent(self) -> Result<(), BoxError> {
184+
if tokio::runtime::Handle::try_current().is_err() {
185+
panic!("`run_concurrent` must be called from within a Tokio runtime");
186+
}
187+
170188
if self.concurrency_limit > 1 {
171189
trace!("Concurrent mode: _X_AMZN_TRACE_ID is not set; use context.xray_trace_id");
172190
Self::run_concurrent_inner(self.service, self.config, self.client, self.concurrency_limit).await
@@ -259,20 +277,20 @@ where
259277
}
260278
}
261279

262-
#[cfg(feature = "experimental-concurrency")]
280+
#[cfg(feature = "concurrency-tokio")]
263281
#[derive(Debug)]
264282
enum WorkerError {
265283
CleanExit(tokio::task::Id),
266284
Failure(tokio::task::Id, BoxError),
267285
}
268286

269-
#[cfg(feature = "experimental-concurrency")]
287+
#[cfg(feature = "concurrency-tokio")]
270288
#[derive(Debug)]
271289
struct ConcurrentWorkerErrors {
272290
errors: Vec<WorkerError>,
273291
}
274292

275-
#[cfg(feature = "experimental-concurrency")]
293+
#[cfg(feature = "concurrency-tokio")]
276294
#[derive(Serialize)]
277295
struct ConcurrentWorkerErrorsPayload<'a> {
278296
message: &'a str,
@@ -282,14 +300,14 @@ struct ConcurrentWorkerErrorsPayload<'a> {
282300
failures: Vec<WorkerFailurePayload>,
283301
}
284302

285-
#[cfg(feature = "experimental-concurrency")]
303+
#[cfg(feature = "concurrency-tokio")]
286304
#[derive(Serialize)]
287305
struct WorkerFailurePayload {
288306
id: String,
289307
err: String,
290308
}
291309

292-
#[cfg(feature = "experimental-concurrency")]
310+
#[cfg(feature = "concurrency-tokio")]
293311
impl fmt::Display for ConcurrentWorkerErrors {
294312
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
295313
let mut clean = Vec::new();
@@ -326,7 +344,7 @@ impl fmt::Display for ConcurrentWorkerErrors {
326344
}
327345
}
328346

329-
#[cfg(feature = "experimental-concurrency")]
347+
#[cfg(feature = "concurrency-tokio")]
330348
impl std::error::Error for ConcurrentWorkerErrors {}
331349

332350
impl<S> Runtime<S>
@@ -335,14 +353,20 @@ where
335353
{
336354
/// Start the runtime and begin polling for events on the Lambda Runtime API.
337355
///
338-
/// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, this returns an error because it does not enable
339-
/// concurrent polling. Enable the `experimental-concurrency` feature and use
340-
/// [`Runtime::run_concurrent`] instead.
356+
/// The runtime will process requests sequentially.
357+
///
358+
/// # Managed concurrency
359+
/// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, a warning is logged.
360+
/// If your handler can satisfy `Clone + Send + 'static`,
361+
/// prefer [`Runtime::run_concurrent`] (requires the `concurrency-tokio` feature),
362+
/// which honors managed concurrency and falls back to sequential behavior when
363+
/// unset.
341364
pub async fn run(self) -> Result<(), BoxError> {
342365
if let Some(raw) = concurrency_env_value() {
343-
return Err(Box::new(io::Error::other(format!(
344-
"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"
345-
))));
366+
tracing::warn!(
367+
"AWS_LAMBDA_MAX_CONCURRENCY is set to '{}', but the concurrency-tokio feature is not enabled; running sequentially",
368+
raw
369+
);
346370
}
347371
let incoming = incoming(&self.client);
348372
Self::run_with_incoming(self.service, self.config, incoming).await
@@ -412,7 +436,7 @@ fn incoming(
412436
}
413437

414438
/// Creates a future that polls the `/next` endpoint.
415-
#[cfg(feature = "experimental-concurrency")]
439+
#[cfg(feature = "concurrency-tokio")]
416440
async fn next_event_future(client: &ApiClient) -> Result<http::Response<hyper::body::Incoming>, BoxError> {
417441
let req = NextEventRequest.into_req()?;
418442
client.call(req).await
@@ -429,7 +453,7 @@ fn concurrency_env_value() -> Option<String> {
429453
env::var("AWS_LAMBDA_MAX_CONCURRENCY").ok()
430454
}
431455

432-
#[cfg(feature = "experimental-concurrency")]
456+
#[cfg(feature = "concurrency-tokio")]
433457
async fn concurrent_worker_loop<S>(mut service: S, config: Arc<Config>, client: Arc<ApiClient>) -> Result<(), BoxError>
434458
where
435459
S: Service<LambdaInvocation, Response = (), Error = BoxError>,
@@ -760,7 +784,7 @@ mod endpoint_tests {
760784
.await
761785
}
762786

763-
#[cfg(feature = "experimental-concurrency")]
787+
#[cfg(feature = "concurrency-tokio")]
764788
#[tokio::test]
765789
async fn concurrent_worker_crash_does_not_stop_other_workers() -> Result<(), Error> {
766790
let next_calls = Arc::new(AtomicUsize::new(0));
@@ -910,7 +934,7 @@ mod endpoint_tests {
910934
}
911935

912936
#[tokio::test]
913-
#[cfg(feature = "experimental-concurrency")]
937+
#[cfg(feature = "concurrency-tokio")]
914938
async fn test_concurrent_structured_logging_isolation() -> Result<(), Error> {
915939
use std::collections::HashSet;
916940
use tracing::info;

0 commit comments

Comments
 (0)