diff --git a/localstack/config.py b/localstack/config.py
index c98a9f6f8d264..cabb9bb72eed1 100644
--- a/localstack/config.py
+++ b/localstack/config.py
@@ -763,9 +763,9 @@ def populate_edge_configuration(
 INTERNAL_RESOURCE_ACCOUNT = os.environ.get("INTERNAL_RESOURCE_ACCOUNT") or "949334387222"
 
 # Determine which implementation to use for the event rule / event filtering engine used by multiple services:
-# EventBridge, EventBridge Pipes, Lambda Event Source Mapping, SNS
-# Options: provider (default) | java
-EVENT_RULE_ENGINE = os.environ.get("EVENT_RULE_ENGINE", "").strip()
+# EventBridge, EventBridge Pipes, Lambda Event Source Mapping
+# Options: python (default) | java (preview)
+EVENT_RULE_ENGINE = os.environ.get("EVENT_RULE_ENGINE", "python").strip()
 
 # -----
 # SERVICE-SPECIFIC CONFIGS BELOW
diff --git a/localstack/services/lambda_/invocation/docker_runtime_executor.py b/localstack/services/lambda_/invocation/docker_runtime_executor.py
index 2240d4329b5b3..aa66d6ceea804 100644
--- a/localstack/services/lambda_/invocation/docker_runtime_executor.py
+++ b/localstack/services/lambda_/invocation/docker_runtime_executor.py
@@ -353,7 +353,8 @@ def start(self, env_vars: dict[str, str]) -> None:
             container_config.dns = config.LAMBDA_DOCKER_DNS
         else:
             if dns_server.is_server_running():
-                # Enable transparent endpoint injection by setting DNS to the embedded DNS re-writer in the Go init.
+                # Set the container DNS to LocalStack to resolve localhost.localstack.cloud and
+                # enable transparent endpoint injection (Pro image only).
                 container_config.dns = self.get_endpoint_from_executor()
 
         lambda_hooks.start_docker_executor.run(container_config, self.function_version)
diff --git a/localstack/services/lambda_/invocation/event_manager.py b/localstack/services/lambda_/invocation/event_manager.py
index 6fc833c8bbd5e..4fc51eefb1703 100644
--- a/localstack/services/lambda_/invocation/event_manager.py
+++ b/localstack/services/lambda_/invocation/event_manager.py
@@ -470,11 +470,14 @@ def start(self) -> None:
             function_id = self.version_manager.function_version.id
             # Truncate function name to ensure queue name limit of max 80 characters
             function_name_short = function_id.function_name[:47]
-            queue_name = f"{function_name_short}-{md5(function_id.qualified_arn())}"
+            # The instance id MUST be unique to the function and a given LocalStack instance
+            queue_namespace = (
+                f"{function_id.qualified_arn()}-{self.version_manager.function.instance_id}"
+            )
+            queue_name = f"{function_name_short}-{md5(queue_namespace)}"
             create_queue_response = sqs_client.create_queue(QueueName=queue_name)
             self.event_queue_url = create_queue_response["QueueUrl"]
-            # Ensure no events are in new queues due to persistence and cloud pods
-            sqs_client.purge_queue(QueueUrl=self.event_queue_url)
+            # We don't need to purge the queue for persistence or cloud pods because the instance id is MUST be unique
 
             self.poller = Poller(self.version_manager, self.event_queue_url)
             self.poller_thread = FuncThread(
diff --git a/localstack/services/lambda_/invocation/lambda_models.py b/localstack/services/lambda_/invocation/lambda_models.py
index 6a9bde3bafdb9..41737340a3b32 100644
--- a/localstack/services/lambda_/invocation/lambda_models.py
+++ b/localstack/services/lambda_/invocation/lambda_models.py
@@ -42,7 +42,7 @@
 from localstack.constants import AWS_REGION_US_EAST_1
 from localstack.services.lambda_.api_utils import qualified_lambda_arn, unqualified_lambda_arn
 from localstack.utils.archives import unzip
-from localstack.utils.strings import long_uid
+from localstack.utils.strings import long_uid, short_uid
 
 LOG = logging.getLogger(__name__)
 
@@ -601,6 +601,25 @@ class Function:
     def latest(self) -> FunctionVersion:
         return self.versions["$LATEST"]
 
+    # HACK to model a volatile variable that should be ignored for persistence
+    def __post_init__(self):
+        # Identifier unique to this function and LocalStack instance.
+        # A LocalStack restart or persistence load should create a new instance id.
+        # Used for retaining invoke queues across version updates for $LATEST, but separate unrelated instances.
+        self.instance_id = short_uid()
+
+    def __getstate__(self):
+        """Ignore certain volatile fields for pickling.
+        # https://docs.python.org/3/library/pickle.html#handling-stateful-objects
+        """
+        # Copy the object's state from self.__dict__ which contains
+        # all our instance attributes. Always use the dict.copy()
+        # method to avoid modifying the original state.
+        state = self.__dict__.copy()
+        # Remove the volatile entries.
+        del state["instance_id"]
+        return state
+
 
 class ValidationException(CommonServiceException):
     def __init__(self, message: str):
diff --git a/localstack/services/lambda_/invocation/version_manager.py b/localstack/services/lambda_/invocation/version_manager.py
index 2ea29001420bf..6a8e265ff39ec 100644
--- a/localstack/services/lambda_/invocation/version_manager.py
+++ b/localstack/services/lambda_/invocation/version_manager.py
@@ -3,7 +3,6 @@
 import threading
 import time
 from concurrent.futures import Future
-from typing import TYPE_CHECKING
 
 from localstack import config
 from localstack.aws.api.lambda_ import (
@@ -33,9 +32,6 @@
 from localstack.utils.strings import long_uid, truncate
 from localstack.utils.threads import FuncThread, start_thread
 
-if TYPE_CHECKING:
-    from localstack.services.lambda_.invocation.lambda_service import LambdaService
-
 LOG = logging.getLogger(__name__)
 
 
@@ -53,8 +49,6 @@ class LambdaVersionManager:
     state: VersionState | None
     provisioned_state: ProvisionedConcurrencyState | None  # TODO: remove?
     log_handler: LogHandler
-    # TODO not sure about this backlink, maybe a callback is better?
-    lambda_service: "LambdaService"
     counting_service: CountingService
     assignment_service: AssignmentService
 
diff --git a/localstack/services/lambda_/provider.py b/localstack/services/lambda_/provider.py
index 6cc56acde0f56..acb36742efaa8 100644
--- a/localstack/services/lambda_/provider.py
+++ b/localstack/services/lambda_/provider.py
@@ -1277,6 +1277,8 @@ def delete_function(
                 destroy_code_if_not_used(code=version.config.code, function=function)
         else:
             # delete the whole function
+            # TODO: introduce locking for safe deletion: We could create a new version at the API layer before
+            #  the old version gets cleaned up in the internal lambda service.
             function = store.functions.pop(function_name)
             for version in function.versions.values():
                 self.lambda_service.stop_version(qualified_arn=version.id.qualified_arn())
diff --git a/tests/aws/services/lambda_/functions/lambda_s3_integration_function_version.py b/tests/aws/services/lambda_/functions/lambda_s3_integration_function_version.py
new file mode 100644
index 0000000000000..3dfc62dda49d1
--- /dev/null
+++ b/tests/aws/services/lambda_/functions/lambda_s3_integration_function_version.py
@@ -0,0 +1,34 @@
+import json
+import os
+import time
+
+import boto3
+
+s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
+S3_BUCKET_NAME = os.environ.get("S3_BUCKET_NAME")
+# Configurable identifier to test function updates
+FUNCTION_VARIANT = os.environ.get("FUNCTION_VARIANT")
+
+
+def handler(event, context):
+    sleep_duration = int(event.get("sleep_seconds", 0))
+    if sleep_duration > 0:
+        print(f"Sleeping for {sleep_duration} seconds ...")
+        time.sleep(sleep_duration)
+        print("... done sleeping")
+
+    request_prefix = event.get("request_prefix")
+    response = {
+        "function_version": context.function_version,
+        "request_id": context.aws_request_id,
+        "request_prefix": request_prefix,
+        "function_variant": FUNCTION_VARIANT,
+    }
+
+    # The side effect is required to test async invokes
+    if S3_BUCKET_NAME:
+        s3_key = f"{request_prefix}--{FUNCTION_VARIANT}"
+        response["s3_key"] = s3_key
+        s3.put_object(Bucket=S3_BUCKET_NAME, Key=s3_key, Body=json.dumps(response))
+
+    return response
diff --git a/tests/aws/services/lambda_/test_lambda.py b/tests/aws/services/lambda_/test_lambda.py
index 6d90a4c84acba..1f78a2e0f462b 100644
--- a/tests/aws/services/lambda_/test_lambda.py
+++ b/tests/aws/services/lambda_/test_lambda.py
@@ -41,6 +41,7 @@
 from localstack.utils.strings import short_uid, to_bytes, to_str
 from localstack.utils.sync import retry, wait_until
 from localstack.utils.testutil import create_lambda_archive
+from tests.aws.services.lambda_.utils import get_s3_keys
 
 LOG = logging.getLogger(__name__)
 
@@ -74,6 +75,9 @@
 TEST_LAMBDA_AWS_PROXY = os.path.join(THIS_FOLDER, "functions/lambda_aws_proxy.py")
 TEST_LAMBDA_AWS_PROXY_FORMAT = os.path.join(THIS_FOLDER, "functions/lambda_aws_proxy_format.py")
 TEST_LAMBDA_PYTHON_S3_INTEGRATION = os.path.join(THIS_FOLDER, "functions/lambda_s3_integration.py")
+TEST_LAMBDA_PYTHON_S3_INTEGRATION_FUNCTION_VERSION = os.path.join(
+    THIS_FOLDER, "functions/lambda_s3_integration_function_version.py"
+)
 TEST_LAMBDA_INTEGRATION_NODEJS = os.path.join(THIS_FOLDER, "functions/lambda_integration.js")
 TEST_LAMBDA_NODEJS = os.path.join(THIS_FOLDER, "functions/lambda_handler.js")
 TEST_LAMBDA_NODEJS_ES6 = os.path.join(THIS_FOLDER, "functions/lambda_handler_es6.mjs")
@@ -540,12 +544,10 @@ def test_mixed_architecture(self, create_lambda_function, aws_client, snapshot):
         payload = json.load(invoke_result_x86["Payload"])
         assert payload.get("platform_machine") == "x86_64"
 
-        update_function_configuration_response = aws_client.lambda_.update_function_code(
+        update_function_code_response = aws_client.lambda_.update_function_code(
             FunctionName=func_name, ZipFile=zip_file, Architectures=[Architecture.arm64]
         )
-        snapshot.match(
-            "update_function_configuration_response", update_function_configuration_response
-        )
+        snapshot.match("update_function_code_response", update_function_code_response)
         aws_client.lambda_.get_waiter(waiter_name="function_updated_v2").wait(
             FunctionName=func_name
         )
@@ -1778,6 +1780,49 @@ def _invoke_function():
 
         assert not errored
 
+    @markers.aws.validated
+    def test_recreate_function(
+        self, aws_client, create_lambda_function, check_lambda_logs, snapshot
+    ):
+        """Recreating a function with the same name should not cause any resource cleanup issues or namespace collisions
+        Reproduces a GitHub issue: https://github.com/localstack/localstack/issues/10280"""
+        function_name = f"test-function-{short_uid()}"
+        create_function_response_one = create_lambda_function(
+            func_name=function_name,
+            handler_file=TEST_LAMBDA_PYTHON_ECHO,
+            handler="lambda_echo.handler",
+            runtime=Runtime.python3_12,
+        )
+        snapshot.match("create_function_response_one", create_function_response_one)
+
+        aws_client.lambda_.delete_function(FunctionName=function_name)
+
+        # Immediately re-create the same function
+        create_function_response_two = create_lambda_function(
+            func_name=function_name,
+            handler_file=TEST_LAMBDA_PYTHON_ECHO,
+            handler="lambda_echo.handler",
+            runtime=Runtime.python3_12,
+        )
+        snapshot.match("create_function_response_two", create_function_response_two)
+
+        # Validate that async invokes still work
+        invoke_response = aws_client.lambda_.invoke(
+            FunctionName=function_name,
+            InvocationType="Event",
+        )
+        invoke_response = read_streams(invoke_response)
+        assert 202 == invoke_response["StatusCode"]
+
+        # Assert that the function gets invoked by checking the logs.
+        # This also ensures that we wait until the invocation is done before deleting the function.
+        expected = [".*{}"]
+
+        def check_logs():
+            check_lambda_logs(function_name, expected_lines=expected)
+
+        retry(check_logs, retries=15)
+
 
 class TestLambdaMultiAccounts:
     @pytest.fixture
@@ -2602,6 +2647,154 @@ def test_lambda_handler_update(self, aws_client, create_lambda_function, snapsho
             "invoke_result_handler_two_postpublish", invoke_result_handler_two_postpublish
         )
 
+    # TODO: Fix this test by not stopping running invokes for function updates of $LATEST
+    @pytest.mark.skip(
+        reason="""Fails with 'Internal error while executing lambda' because
+                  the current implementation stops all running invokes upon update."""
+    )
+    @markers.aws.validated
+    def test_function_update_during_invoke(self, aws_client, create_lambda_function, snapshot):
+        function_name = f"test-function-{short_uid()}"
+        environment_v1 = {"Variables": {"FUNCTION_VARIANT": "variant-1"}}
+        create_lambda_function(
+            func_name=function_name,
+            handler_file=TEST_LAMBDA_PYTHON_S3_INTEGRATION_FUNCTION_VERSION,
+            runtime=Runtime.python3_12,
+            Environment=environment_v1,
+        )
+
+        errored = False
+
+        def _update_function():
+            nonlocal errored
+            try:
+                # Make it very likely that the invocation is being processed because the incoming invocation acquires
+                # an invocation lease quickly.
+                time.sleep(5)
+
+                environment_v2 = environment_v1.copy()
+                environment_v2["Variables"]["FUNCTION_VARIANT"] = "variant-2"
+                aws_client.lambda_.update_function_configuration(
+                    FunctionName=function_name, Environment=environment_v2
+                )
+                waiter = aws_client.lambda_.get_waiter("function_updated_v2")
+                waiter.wait(FunctionName=function_name)
+
+                payload = {"request_prefix": "2-post-update"}
+                invoke_response_after = aws_client.lambda_.invoke(
+                    FunctionName=function_name,
+                    Payload=json.dumps(payload),
+                )
+                assert invoke_response_after["StatusCode"] == 200
+                payload = json.load(invoke_response_after["Payload"])
+                assert payload["function_variant"] == "variant-2"
+                assert payload["function_version"] == "$LATEST"
+            except Exception:
+                LOG.exception(f"Updating lambda function {function_name} failed.")
+                errored = True
+
+        # Start thread with upcoming function update (slightly delayed)
+        thread = threading.Thread(target=_update_function)
+        thread.start()
+
+        # Start an invocation with a sleep
+        payload = {"request_prefix": "1-sleep", "sleep_seconds": 20}
+        invoke_response_before = aws_client.lambda_.invoke(
+            FunctionName=function_name,
+            Payload=json.dumps(payload),
+        )
+        snapshot.match("invoke_response_before", invoke_response_before)
+
+        thread.join()
+        assert not errored
+
+    # TODO: Fix first invoke getting retried and ending up being executed against the new variant because the
+    #  update stops the running function version. We should let running executions finish for $LATEST in this case.
+    # MAYBE: consider validating whether a code update behaves differently than a configuration update
+    @markers.aws.validated
+    def test_async_invoke_queue_upon_function_update(
+        self, aws_client, create_lambda_function, s3_create_bucket, snapshot
+    ):
+        """Test what happens with queued async invokes (i.e., event invokes) when updating a function.
+        We are using a combination of reserved concurrency and sleeps to design this test case predictable.
+        Observation: If we don't wait after sending the first invoke, some queued invokes can still be handled by an
+        old variant in some non-deterministic way.
+        """
+        # HACK: workaround to ignore `$..async_invoke_history_sorted[0]` because indices don't work in the ignore list
+        snapshot.add_transformer(
+            snapshot.transform.regex("01-sleep--variant-2", "01-sleep--variant-1")
+        )
+        bucket_name = f"lambda-target-bucket-{short_uid()}"
+        s3_create_bucket(Bucket=bucket_name)
+
+        function_name = f"test-function-{short_uid()}"
+        environment_v1 = {
+            "Variables": {"S3_BUCKET_NAME": bucket_name, "FUNCTION_VARIANT": "variant-1"}
+        }
+        create_lambda_function(
+            func_name=function_name,
+            handler_file=TEST_LAMBDA_PYTHON_S3_INTEGRATION_FUNCTION_VERSION,
+            runtime=Runtime.python3_12,
+            Environment=environment_v1,
+        )
+        # Add reserved concurrency limits the throughput and makes it easier to cause event invokes to queue up.
+        reserved_concurrency_response = aws_client.lambda_.put_function_concurrency(
+            FunctionName=function_name,
+            ReservedConcurrentExecutions=1,
+        )
+        assert reserved_concurrency_response["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+        payload = {"request_prefix": f"{1:02}-sleep", "sleep_seconds": 22}
+        aws_client.lambda_.invoke(
+            FunctionName=function_name,
+            InvocationType="Event",
+            Payload=json.dumps(payload),
+        )
+        # Make it very likely that the invocation is being processed because the Lambda poller should pick up queued
+        # async invokes quickly using long polling.
+        time.sleep(2)
+
+        # Send async invocation, which should queue up before we update the function
+        num_invocations_before = 9
+        for index in range(num_invocations_before):
+            payload = {"request_prefix": f"{index + 2:02}-before"}
+            aws_client.lambda_.invoke(
+                FunctionName=function_name,
+                InvocationType="Event",
+                Payload=json.dumps(payload),
+            )
+
+        # Update the function variant while still having invokes in the async invoke queue
+        environment_v2 = environment_v1.copy()
+        environment_v2["Variables"]["FUNCTION_VARIANT"] = "variant-2"
+        aws_client.lambda_.update_function_configuration(
+            FunctionName=function_name, Environment=environment_v2
+        )
+        waiter = aws_client.lambda_.get_waiter("function_updated_v2")
+        waiter.wait(FunctionName=function_name)
+
+        # Send further async invocations after the update succeeded
+        num_invocations_after = 5
+        for index in range(num_invocations_after):
+            payload = {"request_prefix": f"{index + num_invocations_before + 2:02}-after"}
+            aws_client.lambda_.invoke(
+                FunctionName=function_name,
+                InvocationType="Event",
+                Payload=json.dumps(payload),
+            )
+
+        # +1 for the first sleep invocation
+        total_invocations = 1 + num_invocations_before + num_invocations_after
+
+        def assert_s3_objects():
+            s3_keys_output = get_s3_keys(aws_client, bucket_name)
+            assert len(s3_keys_output) >= total_invocations
+            return s3_keys_output
+
+        s3_keys = retry(assert_s3_objects, retries=20, sleep=5)
+        s3_keys_sorted = sorted(s3_keys)
+        snapshot.match("async_invoke_history_sorted", s3_keys_sorted)
+
 
 # TODO: test if routing is static for a single invocation:
 #  Do retries for an event invoke, take the same "path" for every retry?
diff --git a/tests/aws/services/lambda_/test_lambda.snapshot.json b/tests/aws/services/lambda_/test_lambda.snapshot.json
index b3a86ad228c54..0e4781d0b78d6 100644
--- a/tests/aws/services/lambda_/test_lambda.snapshot.json
+++ b/tests/aws/services/lambda_/test_lambda.snapshot.json
@@ -3522,7 +3522,7 @@
     "recorded-content": {}
   },
   "tests/aws/services/lambda_/test_lambda.py::TestLambdaBehavior::test_mixed_architecture": {
-    "recorded-date": "08-04-2024, 16:55:53",
+    "recorded-date": "15-05-2024, 12:55:53",
     "recorded-content": {
       "create_function_response": {
         "CreateEventSourceMappingResponse": null,
@@ -3573,7 +3573,7 @@
           }
         }
       },
-      "update_function_configuration_response": {
+      "update_function_code_response": {
         "Architectures": [
           "arm64"
         ],
@@ -4161,5 +4161,149 @@
   "tests/aws/services/lambda_/test_lambda.py::TestLambdaMultiAccounts::test_cross_account_access": {
     "recorded-date": "08-04-2024, 16:59:40",
     "recorded-content": {}
+  },
+  "tests/aws/services/lambda_/test_lambda.py::TestLambdaCleanup::test_recreate_function": {
+    "recorded-date": "15-05-2024, 10:16:45",
+    "recorded-content": {
+      "create_function_response_one": {
+        "CreateEventSourceMappingResponse": null,
+        "CreateFunctionResponse": {
+          "Architectures": [
+            "x86_64"
+          ],
+          "CodeSha256": "code-sha256",
+          "CodeSize": "<code-size>",
+          "Description": "",
+          "Environment": {
+            "Variables": {}
+          },
+          "EphemeralStorage": {
+            "Size": 512
+          },
+          "FunctionArn": "arn:aws:lambda:<region>:111111111111:function:<function-name:1>",
+          "FunctionName": "<function-name:1>",
+          "Handler": "lambda_echo.handler",
+          "LastModified": "date",
+          "LoggingConfig": {
+            "LogFormat": "Text",
+            "LogGroup": "/aws/lambda/<function-name:1>"
+          },
+          "MemorySize": 128,
+          "PackageType": "Zip",
+          "RevisionId": "<uuid:1>",
+          "Role": "arn:aws:iam::111111111111:role/<resource:1>",
+          "Runtime": "python3.12",
+          "RuntimeVersionConfig": {
+            "RuntimeVersionArn": "arn:aws:lambda:<region>::runtime:<resource:2>"
+          },
+          "SnapStart": {
+            "ApplyOn": "None",
+            "OptimizationStatus": "Off"
+          },
+          "State": "Pending",
+          "StateReason": "The function is being created.",
+          "StateReasonCode": "Creating",
+          "Timeout": 30,
+          "TracingConfig": {
+            "Mode": "PassThrough"
+          },
+          "Version": "$LATEST",
+          "ResponseMetadata": {
+            "HTTPHeaders": {},
+            "HTTPStatusCode": 201
+          }
+        }
+      },
+      "create_function_response_two": {
+        "CreateEventSourceMappingResponse": null,
+        "CreateFunctionResponse": {
+          "Architectures": [
+            "x86_64"
+          ],
+          "CodeSha256": "code-sha256",
+          "CodeSize": "<code-size>",
+          "Description": "",
+          "Environment": {
+            "Variables": {}
+          },
+          "EphemeralStorage": {
+            "Size": 512
+          },
+          "FunctionArn": "arn:aws:lambda:<region>:111111111111:function:<function-name:1>",
+          "FunctionName": "<function-name:1>",
+          "Handler": "lambda_echo.handler",
+          "LastModified": "date",
+          "LoggingConfig": {
+            "LogFormat": "Text",
+            "LogGroup": "/aws/lambda/<function-name:1>"
+          },
+          "MemorySize": 128,
+          "PackageType": "Zip",
+          "RevisionId": "<uuid:2>",
+          "Role": "arn:aws:iam::111111111111:role/<resource:1>",
+          "Runtime": "python3.12",
+          "RuntimeVersionConfig": {
+            "RuntimeVersionArn": "arn:aws:lambda:<region>::runtime:<resource:2>"
+          },
+          "SnapStart": {
+            "ApplyOn": "None",
+            "OptimizationStatus": "Off"
+          },
+          "State": "Pending",
+          "StateReason": "The function is being created.",
+          "StateReasonCode": "Creating",
+          "Timeout": 30,
+          "TracingConfig": {
+            "Mode": "PassThrough"
+          },
+          "Version": "$LATEST",
+          "ResponseMetadata": {
+            "HTTPHeaders": {},
+            "HTTPStatusCode": 201
+          }
+        }
+      }
+    }
+  },
+  "tests/aws/services/lambda_/test_lambda.py::TestLambdaVersions::test_async_invoke_queue_upon_function_update": {
+    "recorded-date": "15-05-2024, 17:38:05",
+    "recorded-content": {
+      "async_invoke_history_sorted": [
+        "01-sleep--variant-1",
+        "02-before--variant-2",
+        "03-before--variant-2",
+        "04-before--variant-2",
+        "05-before--variant-2",
+        "06-before--variant-2",
+        "07-before--variant-2",
+        "08-before--variant-2",
+        "09-before--variant-2",
+        "10-before--variant-2",
+        "11-after--variant-2",
+        "12-after--variant-2",
+        "13-after--variant-2",
+        "14-after--variant-2",
+        "15-after--variant-2"
+      ]
+    }
+  },
+  "tests/aws/services/lambda_/test_lambda.py::TestLambdaVersions::test_function_update_during_invoke": {
+    "recorded-date": "15-05-2024, 19:05:05",
+    "recorded-content": {
+      "invoke_response_before": {
+        "ExecutedVersion": "$LATEST",
+        "Payload": {
+          "function_version": "$LATEST",
+          "request_id": "<uuid:1>",
+          "request_prefix": "1-sleep",
+          "function_variant": "variant-1"
+        },
+        "StatusCode": 200,
+        "ResponseMetadata": {
+          "HTTPHeaders": {},
+          "HTTPStatusCode": 200
+        }
+      }
+    }
   }
 }
diff --git a/tests/aws/services/lambda_/test_lambda.validation.json b/tests/aws/services/lambda_/test_lambda.validation.json
index 3841acdb6dead..1494b0148784b 100644
--- a/tests/aws/services/lambda_/test_lambda.validation.json
+++ b/tests/aws/services/lambda_/test_lambda.validation.json
@@ -45,7 +45,7 @@
     "last_validated_date": "2024-04-08T16:56:04+00:00"
   },
   "tests/aws/services/lambda_/test_lambda.py::TestLambdaBehavior::test_mixed_architecture": {
-    "last_validated_date": "2024-04-08T16:55:52+00:00"
+    "last_validated_date": "2024-05-15T12:55:52+00:00"
   },
   "tests/aws/services/lambda_/test_lambda.py::TestLambdaBehavior::test_runtime_introspection_arm": {
     "last_validated_date": "2024-04-08T16:55:44+00:00"
@@ -56,6 +56,9 @@
   "tests/aws/services/lambda_/test_lambda.py::TestLambdaBehavior::test_runtime_ulimits": {
     "last_validated_date": "2024-04-16T08:12:11+00:00"
   },
+  "tests/aws/services/lambda_/test_lambda.py::TestLambdaCleanup::test_recreate_function": {
+    "last_validated_date": "2024-05-15T10:16:44+00:00"
+  },
   "tests/aws/services/lambda_/test_lambda.py::TestLambdaConcurrency::test_lambda_concurrency_block": {
     "last_validated_date": "2024-04-08T17:02:06+00:00"
   },
@@ -203,6 +206,12 @@
   "tests/aws/services/lambda_/test_lambda.py::TestLambdaURL::test_lambda_url_non_existing_url": {
     "last_validated_date": "2024-04-11T17:16:39+00:00"
   },
+  "tests/aws/services/lambda_/test_lambda.py::TestLambdaVersions::test_async_invoke_queue_upon_function_update": {
+    "last_validated_date": "2024-05-15T18:29:38+00:00"
+  },
+  "tests/aws/services/lambda_/test_lambda.py::TestLambdaVersions::test_function_update_during_invoke": {
+    "last_validated_date": "2024-05-15T19:05:04+00:00"
+  },
   "tests/aws/services/lambda_/test_lambda.py::TestLambdaVersions::test_lambda_handler_update": {
     "last_validated_date": "2024-04-08T17:10:58+00:00"
   },
diff --git a/tests/aws/services/lambda_/test_lambda_performance.py b/tests/aws/services/lambda_/test_lambda_performance.py
index d5d865636e930..7a0b5959a35ec 100644
--- a/tests/aws/services/lambda_/test_lambda_performance.py
+++ b/tests/aws/services/lambda_/test_lambda_performance.py
@@ -32,6 +32,7 @@
     TEST_LAMBDA_PYTHON_ECHO,
     TEST_LAMBDA_PYTHON_S3_INTEGRATION,
 )
+from tests.aws.services.lambda_.utils import get_s3_keys
 
 # These performance tests are opt-in because we currently do not track performance systematically.
 if not is_env_true("TEST_PERFORMANCE"):
@@ -690,16 +691,6 @@ def assert_s3_objects():
     ]
 
 
-def get_s3_keys(aws_client, s3_bucket) -> [str]:
-    s3_keys_output = []
-    paginator = aws_client.s3.get_paginator("list_objects_v2")
-    page_iterator = paginator.paginate(Bucket=s3_bucket)
-    for page in page_iterator:
-        for obj in page.get("Contents", []):
-            s3_keys_output.append(obj["Key"])
-    return s3_keys_output
-
-
 def format_summary(timings: [float]) -> str:
     """Format summary statistics in seconds."""
     p99 = (
diff --git a/tests/aws/services/lambda_/utils.py b/tests/aws/services/lambda_/utils.py
new file mode 100644
index 0000000000000..1f7aac51ab497
--- /dev/null
+++ b/tests/aws/services/lambda_/utils.py
@@ -0,0 +1,13 @@
+"""Test utils for Lambda."""
+
+from localstack.aws.connect import ServiceLevelClientFactory
+
+
+def get_s3_keys(aws_client: ServiceLevelClientFactory, s3_bucket: str) -> list[str]:
+    s3_keys_output = []
+    paginator = aws_client.s3.get_paginator("list_objects_v2")
+    page_iterator = paginator.paginate(Bucket=s3_bucket)
+    for page in page_iterator:
+        for obj in page.get("Contents", []):
+            s3_keys_output.append(obj["Key"])
+    return s3_keys_output