chore: additional fixes

Author: darklight3it

.github/workflows/dockerized-test.yml

@@ -57,7 +57,7 @@ jobs:
     - name: Build Lambda artifacts for testing
       run: |
         mkdir -p test/dockerized/tasks
-        HANDLERS_TO_BUILD="basic-lambda-concurrent" OUTPUT_DIR="$(pwd)/test/dockerized/tasks" make build-examples
+        HANDLERS_TO_BUILD="basic-lambda-concurrent invocation-id-concurrent" OUTPUT_DIR="$(pwd)/test/dockerized/tasks" make build-examples
         ls -la test/dockerized/tasks/
 
     - name: Build base test image with RIE and custom entrypoint

Makefile

@@ -125,7 +125,7 @@ fmt:
 	cargo +nightly fmt --all
 
 build-examples:
-	HANDLERS_TO_BUILD=${HANDLERS_TO_BUILD} OUTPUT_DIR=${OUTPUT_DIR} ./scripts/build-examples.sh
+	HANDLERS_TO_BUILD="$(HANDLERS_TO_BUILD)" OUTPUT_DIR="$(OUTPUT_DIR)" ./scripts/build-examples.sh
 
 nuke:
 	docker kill $$(docker ps -q)
@@ -145,6 +145,7 @@ build-test-runner: build-examples
 	@echo "Building test runner Docker image..."
 	@docker build -t test-runner:local -f .test-runner/Dockerfile .test-runner
 
+test-dockerized-concurrent: HANDLERS_TO_BUILD := basic-lambda-concurrent invocation-id-concurrent
 test-dockerized-concurrent: build-test-runner
 	@echo "Running concurrent scenarios in Docker..."
 	@docker network rm concurrent-test-net 2>/dev/null || true

examples/invocation-id-concurrent/src/main.rs

@@ -6,8 +6,9 @@ use serde::{Deserialize, Serialize};
 
 #[derive(Deserialize)]
 struct Request {
-    command: String,
-    sleep: u32
+    #[serde(rename = "command")]
+    _command: String,
+    sleep: u32,
 }
 
 #[derive(Serialize, Debug, PartialEq)]

lambda-runtime/src/requests.rs

@@ -138,10 +138,7 @@ where
                 let body = serde_json::to_vec(&body)?;
                 let body = Body::from(body);
 
-                let mut req = build_request()
-                    .method(Method::POST)
-                    .uri(uri)
-                    .body(body)?;
+                let mut req = build_request().method(Method::POST).uri(uri).body(body)?;
 
                 if let Some(id) = self.invocation_id {
                     req.headers_mut().insert(LAMBDA_RUNTIME_INVOCATION_ID, id.parse()?);
@@ -346,8 +343,7 @@ mod tests {
             let stream_response: StreamResponse<_> = stream.into();
             let response = FunctionResponse::StreamingResponse(stream_response);
 
-            let req: EventCompletionRequest<'_, _, (), _, _, _> =
-                EventCompletionRequest::new("id", None, response);
+            let req: EventCompletionRequest<'_, _, (), _, _, _> = EventCompletionRequest::new("id", None, response);
 
             let http_req = req.into_req().expect("into_req should succeed");
 
@@ -466,7 +462,7 @@ mod tests {
                 let stream_response: StreamResponse<_> = stream.into();
                 let response = FunctionResponse::StreamingResponse(stream_response);
 
-                let req: EventCompletionRequest<'_, _, (), _, _, _> = EventCompletionRequest::new("id", response);
+                let req: EventCompletionRequest<'_, _, (), _, _, _> = EventCompletionRequest::new("id", None, response);
 
                 let http_req = req.into_req().expect("into_req should succeed");
 

lambda-runtime/src/runtime.rs

@@ -716,27 +716,15 @@ mod endpoint_tests {
         requests::{EventCompletionRequest, EventErrorRequest, IntoRequest, NextEventRequest},
         BoxFuture, Config, Diagnostic, Error, Runtime,
     };
-    use bytes::Bytes;
-    use http::{HeaderValue, Method, Request, Response, StatusCode};
-    use http_body_util::{BodyExt, Full};
+    use http::{HeaderValue, StatusCode};
+    use http_body_util::BodyExt;
     use httpmock::prelude::*;
 
-    use hyper::{body::Incoming, service::service_fn};
-    use hyper_util::{
-        rt::{tokio::TokioIo, TokioExecutor},
-        server::conn::auto::Builder as ServerBuilder,
-    };
     use lambda_runtime_api_client::PooledClient as Client;
     use std::{
-        convert::Infallible,
         env,
-        sync::{
-            atomic::{AtomicUsize, Ordering},
-            Arc,
-        },
-        time::Duration,
+        sync::Arc,
     };
-    use tokio::{net::TcpListener, sync::Notify};
     use tokio_stream::StreamExt;
 
     #[tokio::test]

lambda-runtime/src/types.rs

@@ -307,7 +307,6 @@ where
 
 #[cfg(test)]
 mod test {
-    use http::HeaderName;
 
     use super::*;
     use crate::Config;
@@ -558,6 +557,8 @@ mod test {
     fn context_with_invocation_id_resolves() {
         let config = Arc::new(Config::default());
         let mut headers = HeaderMap::new();
+        headers.insert("lambda-runtime-aws-request-id", HeaderValue::from_static("my-id"));
+        headers.insert("lambda-runtime-deadline-ms", HeaderValue::from_static("123"));
 
         let context = Context::new("id", config, &headers).unwrap();
 

test/dockerized/scenarios/concurrent_scenarios.py

@@ -9,7 +9,9 @@
 from containerized_test_runner.models import Request, ConcurrentTest
 
 HANDLER = "basic-lambda-concurrent"
+INVOCATION_ID_HANDLER = "invocation-id-concurrent"
 IMAGE = os.environ.get("TEST_IMAGE", "local/test-base")
+SAME_REQUEST_ID = "shared-request-id"
 DEFAULT_CONCURRENCY = 10
 TIMEOUT = 5
 
@@ -23,7 +25,7 @@ def _make_env(concurrency: int = DEFAULT_CONCURRENCY) -> dict:
 
 
 def _invocation_id_env(concurrency: int = DEFAULT_CONCURRENCY, timeout: int = TIMEOUT) -> dict:
-    return  _make_env | {
+    return _make_env(concurrency) | {
         "AWS_LAMBDA_FUNCTION_TIMEOUT": str(timeout),
     }
 
@@ -71,25 +73,25 @@ def get_concurrent_scenarios():
     return scenarios
 
 
-def invocation_id_scenarios():
+def get_invocation_id_scenarios():
     batches = [
         [Request.create(
-            payload={"name": "invoke-A", "sleep": TIMEOUT + 2},
+            payload={"command": "invoke-A", "sleep": TIMEOUT + 2},
             assertions=[{"transform": ".errorType", "error": "Sandbox.Timedout"}],
             headers={"X-Amzn-RequestId": SAME_REQUEST_ID},
         )],
         [Request.create(
-            payload={"name": "invoke-B", "sleep": TIMEOUT - 1},
-            assertions={"response": {"from": "invoke-B"}},
+            payload={"command": "invoke-B", "sleep": TIMEOUT - 1},
+            assertions=[{"transform": ".req_id", "response": SAME_REQUEST_ID}],
             headers={"X-Amzn-RequestId": SAME_REQUEST_ID},
         )],
     ]
 
 
     return [ConcurrentTest(
         name="invocation_id",
-        handler="invocation-id-concurrent",
-        environment_variables=_invocation_id_env(timeout=1),
+        handler=INVOCATION_ID_HANDLER,
+        environment_variables=_invocation_id_env(timeout=TIMEOUT),
         request_batches=batches,
         image=IMAGE,
     )]