lambda_runtime/
lib.rs

1#![deny(clippy::all, clippy::cargo)]
2#![allow(clippy::multiple_crate_versions)]
3#![warn(missing_docs, nonstandard_style, rust_2018_idioms)]
4#![cfg_attr(docsrs, feature(doc_cfg))]
5
6//! The mechanism available for defining a Lambda function is as follows:
7//!
8//! Create a type that conforms to the [`tower::Service`] trait. This type can
9//! then be passed to the the `lambda_runtime::run` function, which launches
10//! and runs the Lambda runtime.
11use serde::{Deserialize, Serialize};
12use std::{
13    env,
14    fmt::{self, Debug},
15    future::Future,
16    pin::Pin,
17    sync::Arc,
18};
19use tokio_stream::Stream;
20use tower::util::ServiceFn;
21pub use tower::{self, service_fn, Service};
22
23#[macro_use]
24mod macros;
25
26/// Diagnostic utilities to convert Rust types into Lambda Error types.
27pub mod diagnostic;
28pub use diagnostic::Diagnostic;
29
30mod deserializer;
31/// Tower middleware to be applied to runtime invocations.
32pub mod layers;
33mod requests;
34mod runtime;
35/// SnapStart snapshot/restore lifecycle support.
36pub mod snapstart;
37/// Utilities for Lambda Streaming functions.
38pub mod streaming;
39
40/// Utilities to initialize and use `tracing` and `tracing-subscriber` in Lambda Functions.
41#[cfg(feature = "tracing")]
42#[cfg_attr(docsrs, doc(cfg(feature = "tracing")))]
43pub use lambda_runtime_api_client::tracing;
44
45/// Types available to a Lambda function.
46mod types;
47
48use requests::EventErrorRequest;
49pub use runtime::{LambdaInvocation, Runtime};
50pub use snapstart::SnapStartResource;
51pub use types::{Context, FunctionResponse, IntoFunctionResponse, LambdaEvent, MetadataPrelude, StreamResponse};
52
53/// A boxed, `Send` future with a bound lifetime — the return type of
54/// [`SnapStartResource`] hooks. Inlined (rather than re-exported from `futures`)
55/// to avoid leaking the `futures` crate into this crate's public type surface.
56pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
57
58/// Error type that lambdas may result in
59pub type Error = lambda_runtime_api_client::BoxError;
60
61/// Configuration derived from environment variables.
62#[derive(Debug, Default, Clone, Eq, PartialEq, Serialize, Deserialize)]
63pub struct Config {
64    /// The name of the function.
65    pub function_name: String,
66    /// The amount of memory available to the function in MB.
67    pub memory: i32,
68    /// The version of the function being executed.
69    pub version: String,
70    /// The name of the Amazon CloudWatch Logs stream for the function.
71    pub log_stream: String,
72    /// The name of the Amazon CloudWatch Logs group for the function.
73    pub log_group: String,
74}
75
76type RefConfig = Arc<Config>;
77
78impl Config {
79    /// Attempts to read configuration from environment variables.
80    pub fn from_env() -> Self {
81        Config {
82            function_name: env::var("AWS_LAMBDA_FUNCTION_NAME").expect("Missing AWS_LAMBDA_FUNCTION_NAME env var"),
83            memory: env::var("AWS_LAMBDA_FUNCTION_MEMORY_SIZE")
84                .expect("Missing AWS_LAMBDA_FUNCTION_MEMORY_SIZE env var")
85                .parse::<i32>()
86                .expect("AWS_LAMBDA_FUNCTION_MEMORY_SIZE env var is not <i32>"),
87            version: env::var("AWS_LAMBDA_FUNCTION_VERSION").expect("Missing AWS_LAMBDA_FUNCTION_VERSION env var"),
88            log_stream: env::var("AWS_LAMBDA_LOG_STREAM_NAME").unwrap_or_default(),
89            log_group: env::var("AWS_LAMBDA_LOG_GROUP_NAME").unwrap_or_default(),
90        }
91    }
92}
93
94/// Return a new [`ServiceFn`] with a closure that takes an event and context as separate arguments.
95#[deprecated(since = "0.5.0", note = "Use `service_fn` and `LambdaEvent` instead")]
96pub fn handler_fn<A, F, Fut>(f: F) -> ServiceFn<impl Fn(LambdaEvent<A>) -> Fut>
97where
98    F: Fn(A, Context) -> Fut,
99{
100    service_fn(move |req: LambdaEvent<A>| f(req.payload, req.context))
101}
102
103/// Starts the Lambda Rust runtime and begins polling for events on the [Lambda
104/// Runtime APIs](https://docs.aws.amazon.com/lambda/latest/dg/runtimes-api.html).
105///
106/// If you need more control over the runtime and add custom middleware, use the
107/// [Runtime] type directly.
108///
109/// # Managed concurrency
110/// If `AWS_LAMBDA_MAX_CONCURRENCY` is set, a warning is logged.
111/// If your handler can satisfy `Clone + Send + 'static`,
112/// prefer [`run_concurrent`] (requires the `concurrency-tokio` feature),
113/// which honors managed concurrency and falls back to sequential behavior when
114/// unset.
115///
116/// # Example
117/// ```no_run
118/// use lambda_runtime::{Error, service_fn, LambdaEvent};
119/// use serde_json::Value;
120///
121/// #[tokio::main]
122/// async fn main() -> Result<(), Error> {
123///     let func = service_fn(func);
124///     lambda_runtime::run(func).await?;
125///     Ok(())
126/// }
127///
128/// async fn func(event: LambdaEvent<Value>) -> Result<Value, Error> {
129///     Ok(event.payload)
130/// }
131/// ```
132///
133/// # Panics
134///
135/// This function panics if required Lambda environment variables are missing
136/// (`AWS_LAMBDA_FUNCTION_NAME`, `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`,
137/// `AWS_LAMBDA_FUNCTION_VERSION`, `AWS_LAMBDA_RUNTIME_API`).
138pub async fn run<A, F, R, B, S, D, E>(handler: F) -> Result<(), Error>
139where
140    F: Service<LambdaEvent<A>, Response = R>,
141    F::Future: Future<Output = Result<R, F::Error>>,
142    F::Error: Into<Diagnostic> + fmt::Debug,
143    A: for<'de> Deserialize<'de>,
144    R: IntoFunctionResponse<B, S>,
145    B: Serialize,
146    S: Stream<Item = Result<D, E>> + Unpin + Send + 'static,
147    D: Into<bytes::Bytes> + Send,
148    E: Into<Error> + Send + Debug,
149{
150    let runtime = Runtime::new(handler).layer(layers::TracingLayer::new());
151    runtime.run().await
152}
153
154/// Starts the Lambda Rust runtime in a mode that is compatible with
155/// Lambda Managed Instances (concurrent invocations).
156///
157/// Requires the `concurrency-tokio` feature.
158///
159/// When `AWS_LAMBDA_MAX_CONCURRENCY` is set to a value greater than 1, this
160/// will spawn `AWS_LAMBDA_MAX_CONCURRENCY` tokio worker tasks, each running its own
161/// `/next` polling loop. When the environment variable is unset or `<= 1`, it
162/// falls back to the same sequential behavior as [`run`], so the same handler
163/// can run on both classic Lambda and Lambda Managed Instances.
164///
165/// If you need more control over the runtime and add custom middleware, use the
166/// [Runtime] type directly.
167///
168/// # Example
169/// ```no_run
170/// use lambda_runtime::{Error, service_fn, LambdaEvent};
171/// use serde_json::Value;
172///
173/// #[tokio::main]
174/// async fn main() -> Result<(), Error> {
175///     let func = service_fn(func);
176///     lambda_runtime::run_concurrent(func).await?;
177///     Ok(())
178/// }
179///
180/// async fn func(event: LambdaEvent<Value>) -> Result<Value, Error> {
181///     Ok(event.payload)
182/// }
183/// ```
184///
185/// # Panics
186///
187/// This function panics if:
188/// - Called outside of a Tokio runtime
189/// - Required Lambda environment variables are missing (`AWS_LAMBDA_FUNCTION_NAME`,
190///   `AWS_LAMBDA_FUNCTION_MEMORY_SIZE`, `AWS_LAMBDA_FUNCTION_VERSION`,
191///   `AWS_LAMBDA_RUNTIME_API`)
192#[cfg(feature = "concurrency-tokio")]
193#[cfg_attr(docsrs, doc(cfg(feature = "concurrency-tokio")))]
194pub async fn run_concurrent<A, F, R, B, S, D, E>(handler: F) -> Result<(), Error>
195where
196    F: Service<LambdaEvent<A>, Response = R> + Clone + Send + 'static,
197    F::Future: Future<Output = Result<R, F::Error>> + Send + 'static,
198    F::Error: Into<Diagnostic> + fmt::Debug,
199    A: for<'de> Deserialize<'de> + Send + 'static,
200    R: IntoFunctionResponse<B, S> + Send + 'static,
201    B: Serialize + Send + 'static,
202    S: Stream<Item = Result<D, E>> + Unpin + Send + 'static,
203    D: Into<bytes::Bytes> + Send + 'static,
204    E: Into<Error> + Send + Debug + 'static,
205{
206    let runtime = Runtime::new(handler).layer(layers::TracingLayer::new());
207    runtime.run_concurrent().await
208}
209
210/// Spawns a task that will be execute a provided async closure when the process
211/// receives unix graceful shutdown signals. If the closure takes longer than 500ms
212/// to execute, an unhandled `SIGKILL` signal might be received.
213///
214/// You can use this future to execute cleanup or flush related logic prior to runtime shutdown.
215///
216/// This function's returned future must be resolved prior to `lambda_runtime::run()`.
217///
218/// Note that this implicitly also registers and drives a no-op internal extension that subscribes to no events.
219/// This extension will be named `_lambda-rust-runtime-no-op-graceful-shutdown-helper`. This extension name
220/// can not be reused by other registered extensions. This is necessary in order to receive graceful shutdown signals.
221///
222/// This extension is cheap to run because it receives no events, but is not zero cost. If you have another extension
223/// registered already, you might prefer to manually construct your own graceful shutdown handling without the dummy extension.
224///
225/// For more information on general AWS Lambda graceful shutdown handling, see:
226/// <https://github.com/aws-samples/graceful-shutdown-with-aws-lambda>
227///
228/// # Panics
229///
230/// This function panics if:
231/// - this function is called after `lambda_runtime::run()`
232/// - this function is called outside of a context that has access to the tokio i/o
233/// - the no-op extension cannot be registered
234/// - either signal listener panics [tokio::signal::unix](https://docs.rs/tokio/latest/tokio/signal/unix/fn.signal.html#errors)
235///
236/// # Example
237/// ```no_run
238/// use lambda_runtime::{Error, service_fn, LambdaEvent};
239/// use serde_json::Value;
240///
241/// #[tokio::main]
242/// async fn main() -> Result<(), Error> {
243///     let func = service_fn(func);
244///
245///     let (writer, log_guard) = tracing_appender::non_blocking(std::io::stdout());
246///     lambda_runtime::tracing::init_default_subscriber_with_writer(writer);
247///
248///     let shutdown_hook = || async move {
249///         std::mem::drop(log_guard);
250///     };
251///     lambda_runtime::spawn_graceful_shutdown_handler(shutdown_hook).await;
252///
253///     lambda_runtime::run(func).await?;
254///     Ok(())
255/// }
256///
257/// async fn func(event: LambdaEvent<Value>) -> Result<Value, Error> {
258///     Ok(event.payload)
259/// }
260/// ```
261#[cfg(all(unix, feature = "graceful-shutdown"))]
262#[cfg_attr(docsrs, doc(cfg(all(unix, feature = "graceful-shutdown"))))]
263pub async fn spawn_graceful_shutdown_handler<Fut>(shutdown_hook: impl FnOnce() -> Fut + Send + 'static)
264where
265    Fut: Future<Output = ()> + Send + 'static,
266{
267    // You need an extension registered with the Lambda orchestrator in order for your process
268    // to receive a SIGTERM for graceful shutdown.
269    //
270    // We accomplish this here by registering a no-op internal extension, which does not subscribe to any events.
271    //
272    // This extension is cheap to run since after it connects to the lambda orchestration, the connection
273    // will just wait forever for data to come, which never comes, so it won't cause wakes.
274    let extension = lambda_extension::Extension::new()
275        // Don't subscribe to any event types
276        .with_events(&[])
277        // Internal extension names MUST be unique within a given Lambda function.
278        .with_extension_name("_lambda-rust-runtime-no-op-graceful-shutdown-helper")
279        // Extensions MUST be registered before calling lambda_runtime::run(), which ends the Init
280        // phase and begins the Invoke phase.
281        .register()
282        .await
283        .expect("could not register no-op extension for graceful shutdown");
284
285    tokio::task::spawn(async move {
286        let graceful_shutdown_future = async move {
287            let mut sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()).unwrap();
288            let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()).unwrap();
289            tokio::select! {
290                _sigint = sigint.recv() => {
291                    eprintln!("[runtime] SIGINT received");
292                    eprintln!("[runtime] Graceful shutdown in progress ...");
293                    shutdown_hook().await;
294                    eprintln!("[runtime] Graceful shutdown completed");
295                    std::process::exit(0);
296                },
297                _sigterm = sigterm.recv()=> {
298                    eprintln!("[runtime] SIGTERM received");
299                    eprintln!("[runtime] Graceful shutdown in progress ...");
300                    shutdown_hook().await;
301                    eprintln!("[runtime] Graceful shutdown completed");
302                    std::process::exit(0);
303                },
304            }
305        };
306
307        let _: (_, ()) = tokio::join!(
308            // we always poll the graceful shutdown future first,
309            // which results in a smaller future due to lack of bookkeeping of which was last polled
310            biased;
311            graceful_shutdown_future, async {
312            // we suppress extension errors because we don't actually mind if it crashes,
313            // all we need to do is kick off the run so that lambda exits the init phase
314            let _ = extension.run().await;
315        });
316    });
317}
lambda_runtime/lib.rs

lambda_runtime/
lib.rs
