Feature: Runtime Execution Simulator

[GradExMachina]

Name Pending.

Testing Lambda code can be difficult / unwieldy. While SAM exists for all supported languages, it is pretty limited in ergonomics, difficult to integrate with in CI, makes testing / infrastructure management unnecessarily complicated, and still requires further orchestration.

In an ideal world, we'd be able to "just" create a runtime with the corresponding service, add all the hooks, tracking, and tracing that we need, execute requests on top, and then verify the results, benchmark, get flamegraphs, etc, all without actually ever deploying, and all from a single tokio::test.

I have added a basic server in #1086, so this backlog issue is for extension / reuse of the implementation.

We should start from an internal implementation, and eventually publish a crate so that users can depend on it and integrate it in their CI.

[jlizen]

Another option would be to publish the simple simulator under a test-util feature. That might be a good first step since a standalone crate presumably might need more work to have a full feature set, more resilience, etc.

We could always cut such a feature over to using a standalone crate once available.

[razx11111]

Name Pending.

Testing Lambda code can be difficult / unwieldy. While SAM exists for all supported languages, it is pretty limited in ergonomics, difficult to integrate with in CI, makes testing / infrastructure management unnecessarily complicated, and still requires further orchestration.

In an ideal world, we'd be able to "just" create a runtime with the corresponding service, add all the hooks, tracking, and tracing that we need, execute requests on top, and then verify the results, benchmark, get flamegraphs, etc, all without actually ever deploying, and all from a single tokio::test.

I have added a basic server in #1086, so this backlog issue is for extension / reuse of the implementation.

We should start from an internal implementation, and eventually publish a crate so that users can depend on it and integrate it in their CI.

Greetings! I'm a comp sci student interested in working on this for my bachelor's thesis(it was already approved by my advisor). Is the approach in PR #1086 still the direction maintainers want to go? Any other guidelines that are not in the docs are warmly welcomed 🙏

[jlizen]

Hi @razx11111 , nice to meet you!

@darklight3it any opinion re:

Is the approach in PR #1086 still the direction maintainers want to go?

I know we've since added some additional simulation tests, wondering how they relate. I'm broadly happy with that approach but open to alternatives. TBH i don't remember much, a bit fuzzy on the proposal.

Also, @GradExMachina are you interested in participating in the reviews on this one too? I know you just started at Datadog / no worries if busy :)

[jlizen]

Given that this is a thesis, I recommend that you cut a internal docs/design-only PR first so we can align on what your plan is, the interfaces, the benefits, etc. That will give you a nice school artifact too.

As part of that, it would be good to consider PR ordering. Is there a minimal starting point that we can ship, that provides some useful functionality that we can evolve? Is there a middle step that is nicer? Is there a golden future that we can use as a stretch goal for your thesis? Are there any users of the runtime that we could integrate with to validate the approach?

A good rule of thumb when designing an API is, find n=2 independent users and make sure the approach works for them. Not strictly required, and it could as simple as some examples/, but the more realistic the better.

[jlizen]

I would also suggest looping in @calavera to discuss whether there is a cargo-lambda integration point that makes the most sense for local usage ergonomics. Probably it's worth cutting a sister issue in their repo if anything comes to mind.

[darklight3it]

Hello everyone!

First, thanks @razx11111 for offering to contribute to this project ❤️ .

@darklight3it any opinion re:

@maxday can correct me, but I think in general the vision for runtimes for testing is to follow the following pillars:

• We should do integration testing consistent across runtimes.
• User should be able to run the test without interacting with resources.
• Testing should be easily exercised by Github Actions.

We want to support 3 layers of testing:

1. Unit tests (in the specific runtime language)
2. Containerized Tests — Test via Docker images. For that we have created this GitHub Action: containerized-test-runner-for-aws-lambda. It spins up the runtime in a container, sends invocations, and asserts on responses — no AWS account needed, no deployed infrastructure, runs entirely in CI.
3. SAM CLI — This layer should only test what cannot be done in the first two layers. Because it relies on deployed resources, it is not expected for contributors to interact with them — only maintainers.

I am not particularly inclined toward an alternative to option 2. The containerized approach gives us the right balance: it exercises the actual runtime binary in the same environment as production (Linux container, same entrypoint), supports concurrent invocations, lets you control resources as needed, and works out of the box in GitHub Actions without credentials or AWS accounts.

A couple of questions:

• Is this intended as a utility for customers to use in their own projects? If so, do you have signal from the community that this is a real need (issues, forum posts, etc.)?
• Is there a specific scenario you see where the containerized approach falls short?

Happy to discuss further ☺️

[GradExMachina]

When I wrote this issue, it was meant as a tool for customers, and for the runtime to test runtime specific stuff. This is meant as an accessory to unit tests. For customers, it's part of their integration tests, without ever spinning a Lambda function, or needing a server running.

Containerized test runners have obvious limitations, especially for rust since rust runtime is a binary that you cannot really instrument to the degree you can dynamically instrument python or node code.

The benefit here is that the runtime logic and loop is exposed as pure code. This means you can programmatically, in your rust tests:

• invoke the function with any arguments you want, as many times as you want, without dealing with sending curl or sam cli requests, all in your integration tests, all executing rust code.
• instrument the function and runtime from within, and as you call your function and do stuff, actively observe internal state changing.

This kind of testing is orders of magnitude more functional for the customer than sam cli, or containerized tests, and makes it significantly easier to unit test the runtime loop itself.

As we have seen multiple times during launches, the containerized tests are garbage at estimating memory used, init time, and all that, and that is a design consequence.

The way to think about this model is that the rapids client gets replaced by a client controlled by the test harness.

Personally, I see significant simplifications and improvements to testing the runtime in this avenue. Ultimately, it's not my call, and I am not really interested in arguing this any further as I am not responsible for maintaining the runtime.

[jlizen]

^ well stated, I broadly agree with Astraea, and think this effort is complimentary to the integration tests enabled by the containerized approach. Happy to review it fwiw, though I would want @darklight3it on board with it being worth having in tree.

[darklight3it]

Hey all!

Sorry for the late response, but I was under the water. Let me give a proper answer.

^ well stated, I broadly agree with Astraea, and think this effort is complimentary to the integration tests enabled by the containerized approach.

I agree that those tests fill a very specific gap in this package but are we sure about giving this tool to the external customers? Customer should care only about their own handlers and marginally cares about the internal state of the runtime.

Do we have any signal from the broader community for this kind of capability? Any specific case?

I don't agree that performance testing is an argument here. We know that docker tests are not a realistic way to perf or profile a function but neither it is an in process test on a github runner.

Also my main concern is that since this kind of tests imply mocking RAPID, aren't we going to reimplement this way what is already implemented in the RIE? If there is a change in RAPID we are sure that this behavior will be reflected in RIE. Are we sure we want to maintain this specifically for this package?

How are we going to explain to customer whether to use a dockerized tests or a this new package?

Not saying no on principle but I would like to know what space are we thinking this new testing package would fill, in your view.

I also agree the ergonomics of containerized tests should be improved. I think every RIC package should have test written in the specific language, using the language most common test runner and their own assertion (rather than json file for harness or even python).

In this case I think we should use what external customers are using. It seems to me that TestContainers is pretty popular and has a rust cargo. We could have integration test with Rust egrgonomics and containers. For example, why not something like that?

use testcontainers::{core::{IntoContainerPort, WaitFor}, runners::AsyncRunner, GenericImage, ImageExt};

#[tokio::test]
async fn test_lmi_basic() {
    let c = GenericImage::new("public.ecr.aws/lambda/provided", "al2023")
        .with_exposed_port(8080.tcp())
        .with_wait_for(WaitFor::message_on_stderr("Ready to accept connections"))
        .with_copy_to("/var/runtime/bootstrap", std::path::Path::new(HANDLER))
        .with_env_var("AWS_LAMBDA_MAX_CONCURRENCY", "10")
        .start().await.unwrap();

    let port = c.get_host_port_ipv4(8080).await.unwrap();
    let url = format!("http://localhost:{port}/2015-03-31/functions/function/invocations");

    let resp: Value = Client::new()
        .post(&url)
        .json(&json!({"command": "hello"}))
        .send().await.unwrap()
        .json().await.unwrap();

    assert_eq!(resp["msg"], "Command hello executed.");
}

[GradExMachina]

The reason for giving that to customers is performance testing and instrumenting their own code, in a loop that they control, without external dependencies other than what this crate provides.

Also my main concern is that since this kind of tests imply mocking RAPID, aren't we going to reimplement this way what is already implemented in the RIE?

No, the client is just an interface. You can implement the client via 2 queues, one to read and one to write.

How are we going to explain to customer whether to use a dockerized tests or a this new package?

When they want to instrument, or don't want to use docker or any container tech in their testing.
A good argument for dockerized is when they have external dependencies that they ship, and extensions.

[darklight3it]

Hey @GradExMachina ,

I think it's fine for me. I discussed also that with @maxday and I think we should go with a more community first approach on testing to help people contribute easier to this package.

As @jlizen was saying I am ok if @razx11111 wants to propose a design for that.

[razx11111]

As @jlizen was saying I am ok if @razx11111 wants to propose a design for that.

Thank you for the greenlight, I will be writing here my ideas so far and only after the design is agreeable and I get to meet with my advisor I will draft a first PR with al the details jlizen suggested

The simulator implements the official AWS Lambda Runtime API HTTP specification rather than mocking internal runtime structs or daemon logic

Queue Implementation Mechanics

Building upon @GradExMachina's idea, the simulator coordinates execution using 2 decoupled channels:

Read Queue (SimulatorEvent): Pushed by the test harness and consumed by GET /2018-06-01/runtime/invocation/next. Using bounded channels provides natural backpressure if events are queued faster than the runtime processes them
Write Queue (SimulatorResponse): Captured by POST /2018-06-01/runtime/invocation/{id}/response (or /error) and pushed back to the test harness

// Internal HTTP Endpoint Handlers

// GET 
 async fn handle_next( event_rx: async_channel::Receiver<SimulatorEvent> ) -> Response<Full<Bytes>> {
    match event_rx.recv().await {

        Ok(event) => Response::builder()
            .status(StatusCode::OK)
            .header("lambda-runtime-aws-request-id", &event.request_id)
            .header("lambda-runtime-deadline-ms", event.deadline_ms.to_string())
            .body(Full::new(event.payload))
            .unwrap(),

        Err(_) => Response::builder() 
            .status(StatusCode::NO_CONTENT)
            .body(Full::default())
            .unwrap(),
    }
}

// POST 
 async fn handle_response( request_id: String, body: Bytes, response_tx: async_channel::Sender<SimulatorResponse> ) -> Response<Full<Bytes>> {
    let _ = response_tx
        .send(SimulatorResponse {
            request_id,
            payload: body,
            is_error: false,
        })
        .await;

    Response::builder()
        .status(StatusCode::OK)
        .body(Full::new(Bytes::from_static(b"{}")))
        .unwrap()
}

Test Util

what writing a test using the simulator would look like:

#[tokio::test]
async fn test_multi_event_handler() -> Result<(), Error> {
    // instantiate simulator & queue events
    let simulator = LambdaSimulator::builder()
        .add_event_with_id("req-1", json!({ "command": "first" }))
        .add_event_with_id("req-2", json!({ "command": "second" }))
        .build()
        .await?;

    // spawn handler against simulator
    let _handle = simulator.spawn(service_fn(my_handler));

    // correlate responses out-of-order by request_id
    let res1 = simulator.response_for("req-1").await?;
    assert_eq!(res1.payload["msg"], "Command first executed.");
    assert!(!res1.is_error);

    let res2 = simulator.response_for("req-2").await?;
    assert_eq!(res2.payload["msg"], "Command second executed.");

    simulator.shutdown().await;
    Ok(())
}

Of course, I'm not really expecting this to be the best option since I'm more limited experience wise. I'll gladly hear any input from you all ❤️

[jlizen]

Is there a reason to require the events to be all wired up in the builder? I would have assumed that the per-request dispatch can live next to the test lines that relate to them.

I guess there might be a mode where you want to shuffle your orderings to avoid improperly handled shared state while using lambda managed instances? In which case having them on the builder would be relevant?

In general I would be interested in seeing more of the public API of the LambdaSimulator, I see a few things in your example, but it might be easier to discuss its method signatures directly?

[razx11111]

"Is there a reason to require the events to be all wired up in the builder? I would have assumed that the per-request dispatch can live next to the test lines that relate to them."

I do agree that requiring events to be pre-wired in the builder would be unnecessary for standard sequential tests. The simulator should support per-request dispatch as the primary workflow but I do consider the builder could be useful for batch configurations.

"I guess there might be a mode where you want to shuffle your orderings to avoid improperly handled shared state while using lambda managed instances? In which case having them on the builder would be relevant?"

the reason to declare events upfront in the builder is specifically for the scenario you described: when testing Lambda Managed Instances with global caches or connection pools, the simulator needs the entire batch upfront so it can randomly shuffle the arrival order across concurrent /next worker polls.

So yes, the design should separate the 2 use cases:

• interactive dispatch for deterministic tests
• builder batching specifically for concurrency and non-deterministic shared-state testing

"In general I would be interested in seeing more of the public API of the LambdaSimulator, I see a few things in your example, but it might be easier to discuss its method signatures directly?"

that would be better, yes. Sorry if the direct use case was too ambiguous

use std::time::Duration; 
use http::Uri;
use bytes::Bytes; 
use serde_json::Value; 
use tower::Service;                                                                                                                                                                    
use lambda_runtime::{LambdaInvocation, BoxError};  

pub struct LambdaSimulator { }                                                                                                                                    
                                                                                                                                                                                           
    impl LambdaSimulator {                                                                                                                                                                 
        /// spawns a new simulator with default settings on an ephemeral port                                                                                                       
        pub async fn start() -> Result<Self, SimulatorError> { }                                                                                                                 
                                                                                                                                                                                           
        /// returns a builder to customize simulator behavior                                                                                    
        pub fn builder() -> SimulatorBuilder { }                                                                                                                                 
                                                                                                                                                                                           
        /// returns the URI of the simulated Runtime API                                                                                      
        pub fn endpoint(&self) -> &Uri { }                                                                                                                                       
                                                                                                                                                                                           
        /// enqueues an event and waits for its matching response                                                                                                       
        /// for linear, single-event unit tests                                                                                                                       
        pub async fn send(&self, event: impl Into<SimulatorEvent>) -> Result<SimulatorResponse, SimulatorError> { }                                                                                
                                                                                                                                                                                          
        /// pushes an event into the queue without blocking for completion                                                                                          
        pub async fn enqueue(&self, event: impl Into<SimulatorEvent>) -> Result<(), SimulatorError> {  }                                                                          
                                                                                                                                                                                           
        /// correlates and awaits a response for a specific `request_id` 
        pub async fn response_for(&self, request_id: &str) -> Result<SimulatorResponse, SimulatorError> { }                               
                                                                                                                                                                                           
        /// spawns the provided Lambda handler as a background task wired to this simulator
        pub fn spawn<A, F, R, B, S, D, E>(&self, handler: F) -> SimulatorHandle                                                                                                            
        where                                                                                                                                                                              
            F: Service<LambdaEvent<A>, Response = R> + Send + 'static,                                                                                                                     
            F::Future: Future<Output = Result<R, F::Error>> + Send + 'static,                                                                                                              
            F::Error: Into<Diagnostic> + std::fmt::Debug,                                                                                                                                  
            A: for<'de> serde::Deserialize<'de> + Send + 'static,                                                                                                                          
            R: IntoFunctionResponse<B, S> + Send + 'static,                                                                                                                                
            B: serde::Serialize + Send + 'static,                                                                                                                                          
            S: tokio_stream::Stream<Item = Result<D, E>> + Unpin + Send + 'static,                                                                                                         
            D: Into<Bytes> + Send + 'static,                                                                                                                                               
            E: Into<BoxError> + Send + std::fmt::Debug + 'static,                                                                                                                          
        { }                                                                                                                                                                  
                                                                                                                                                                                           
        /// gracefully stops the simulator server and underlying tasks                                                                                                      
        pub async fn shutdown(self) -> Result<(), SimulatorError> { }                                                                                                            
    }  

I'll also include the SimulatorBuilder and the adjacent structs

#[derive(Default)]                                                                                                                                                                     
    pub struct SimulatorBuilder { }                                                                                                                                   
                                                                                                                                                                                           
    impl SimulatorBuilder {                                                                                                                                                                
        pub fn new() -> Self { }                                                                                                                                                 
                                                                                                                                                                                           
        /// pre-seeds an event into the invocation queue before startup                                                                                                     
        pub fn add_event(mut self, event: impl Into<SimulatorEvent>) -> Self { }                                                                                                 
                                                                                                                                                                                           
        /// pre-seeds multiple events                                                                                                                                                     
        pub fn add_events(mut self, events: impl IntoIterator<Item = SimulatorEvent>) -> Self {  }                                                                                

        /// enables random shuffling
        pub fn shuffle_order(mut self, shuffle: bool) -> Self { }

        /// seeds the random number generator for reproducible shuffle order in CI
        pub fn with_seed(mut self, seed: u64) -> Self { /* ... */ } 

        /// sets a simulated network delay for runtime API calls
        pub fn with_latency(mut self, latency: Duration) -> Self {  }
                                                                                                                                                                                           
        /// builds and starts the simulator instance
        pub async fn build(self) -> Result<LambdaSimulator, SimulatorError> { }
    }

#[derive(Clone, Debug)]                                                                                                                                                                
    pub struct SimulatorEvent {                                                                                                                                                            
        pub request_id: String,                                                                                                                                                            
        pub deadline_ms: u64,                                                                                                                                                              
        pub xray_trace_id: Option<String>,                                                                                                                                                 
        pub payload: Bytes,                                                                                                                                                                
    }                                                                                                                                                                                      
                                                                                                                                                                                           
    impl SimulatorEvent {                                                                                                                                                                  
        pub fn new(payload: impl Into<Bytes>) -> Self { }                                                                                                      
        pub fn with_id(request_id: impl Into<String>, payload: impl Into<Bytes>) -> Self { 
        pub fn with_deadline_ms(mut self, deadline_ms: u64) -> Self { }
        pub fn with_xray_trace_id(mut self, trace_id: impl Into<String>) -> Self { }
    }                                                                                                                                                                                      
                                                                                                                                                                                           
    impl From<Value> for SimulatorEvent {                                                                                                                                                  
        fn from(val: Value) -> Self {                                                                                                                                                      
            SimulatorEvent::new(serde_json::to_vec(&val).unwrap())                                                                                                                         
        }                                                                                                                                                                                  
    }                                                                                                                                                                                      
                                                                                                                                                                                           
    #[derive(Clone, Debug)]                                                                                                                                                                
    pub struct SimulatorResponse {                                                                                                                                                         
        pub request_id: String,                                                                                                                                                            
        pub payload: Bytes,                                                                                                                                                                
        pub is_error: bool,                                                                                                                                                                
    }                                                                                                                                                                                      
                                                                                                                                                                                           
    impl SimulatorResponse {                                                                                                                                                               
        /// attempts to deserialize the response payload as json                                                                                                                          
        pub fn json<T: serde::de::DeserializeOwned>(&self) -> Result<T, serde_json::Error> {                                                                                               
            serde_json::from_slice(&self.payload)                                                                                                                                          
        }                                                                                                                                                                                  
    }            

pub struct SimulatorHandle { }                                                                                                                                                                                      
                                                                                                                                                                                           
    impl SimulatorHandle {                                                                                                                                                                 
        /// aborts the running handler                                                                                                                                                   
        pub fn abort(&self) {  }                                                                                                                                                  
                                                                                                                                                                                           
        /// awaits handler completion if it exits                                                                                                                                         
        pub async fn join(self) -> Result<(), BoxError> {  }                                                                                                                      
    }                                                                                                                                                                                      
                                                                                                                                                                                           
  #[derive(thiserror::Error, Debug)]                                                                                                                                                     
  pub enum SimulatorError {                                                                                                                                                              
      #[error("I/O or server binding error: {0}")]                                                                                                                                       
      Io(#[from] std::io::Error),                                                                                                                                                        
      #[error("Serialization error: {0}")]                                                                                                                                               
      Serialization(#[from] serde_json::Error),                                                                                                                                          
      #[error("Response timeout for request_id: {0}")]                                                                                                                                   
      Timeout(String),                                                                                                                                                                   
      #[error("Simulator channel closed unexpectedly")]                                                                                                                                  
      ChannelClosed,                                                                                                                                                                     
  }

Sorry if it seems too spammish. I hope the approach is a bit clearer now. Thank you veeeery much for your observations and questions 🙏