Repository: localstack/localstack
Pull request: Add hot reloading for new lambda provider
## Motivation

For efficient lambda development, we have to provide an easy solution to hot reload lambda functions, especially those of interpreted languages (like python and nodejs).
However, we want to make it available for all managed runtimes.
It should also only reset the environments if a change is detected, to prevent unnecessary initializations.

## Breaking Changes with Fallbacks

* Change of the default hot-reloading bucket marker, from `__local__` to `hot-reload`. The current provider will continue to support the `__local__` marker for now, the new one will only support the new one. Would move the deprecation to 1.4 and the removal to 2.0.

## Other Changes

* Add hot-reloading support
* Allow VolumeBind objects to be used directly in docker client, and add option for read only mounts (necessary for hot reloading)
* Add test for hot reloading

Fixes #7200

Problem statement:
feature request: Lambda ASF provider code hot-reload
### Is there an existing issue for this?

- [X] I have searched the existing issues

### Feature description

The new ASF provider should feature the option for hot reloading for all supported languages.
It should only restart environments when necessary, and should be easy to use.

### 🧑‍💻 Implementation

Any volume mounts should be readonly.
Probably a filewatcher on the mounted directory with debounced infrastructure restart in the environments.

### Anything else?

_No response_

Hints from issue discussion:
Can't wait to use this feature. This is the key thing that can turn LocalStack to a truly viable solution for our team.  
I’m not sure if you implement one combined solution for all runtimes, or if it varies based on the language/version, but Node v18 introduced [watch mode](https://nodejs.org/api/cli.html#--watch) which might be a nice way of quickly detecting and restarting the lambda without needing to restart the entire container.
Good hint, thanks! Our solution will however be runtime agnostic, basically we will restart the whole RIC (or whatever custom runtime someone is starting) and all other infrastructure (extensions, etc) within the container, letting it therefore reload.
It will basically be the same kind of reset which happens after a timeout, both in LS and in AWS, the environment is resetted, but not recreated (i.E. local files in /tmp will still be there, any data in the interpreter / process (like global state) will be lost).
It will probably result in some ms delay for the first invocation afterwards, but it should be negligible.
I have some results for this to share!
Python3.9 lambda, results will vary depending on the cold-start time of the specific runtime:
Cold invocation (including container start, with hot reloading): `593.60ms`
Hot invocation (no changes in hot reloading dir): `16.73ms`
Hot invocation (changes since last invocation): `83.94ms`

So we see around `70ms` additional delay here, since we have to restart the python interpreter here.
Currently the interpreter only restarts 500ms after the last file system change was detected (to debounce massive copy operations a bit).
This also means, while file operations are in progress, new code will not be loaded, only 500ms after things settled down a bit.
The path is mounted as read only from the host (to avoid in-lambda changes to trigger a problem).

A reset will result in the lambda returning nothing, but also not failing, for now (so active executions will be cancelled).

It should be able to be tested soon!

Commit message: add developer tools test to circleci config, fix another <3.10 issue

Patch to review:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 4458197d61825..20db836d1decf 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -147,7 +147,7 @@ jobs:
           name: Test ASF Lambda provider
           environment:
             PROVIDER_OVERRIDE_LAMBDA: "asf"
-            TEST_PATH: "tests/integration/awslambda/test_lambda.py tests/integration/awslambda/test_lambda_api.py tests/integration/awslambda/test_lambda_common.py tests/integration/awslambda/test_lambda_integration_sqs.py tests/integration/cloudformation/resources/test_lambda.py tests/integration/awslambda/test_lambda_integration_dynamodbstreams.py tests/integration/awslambda/test_lambda_integration_kinesis.py"
+            TEST_PATH: "tests/integration/awslambda/test_lambda.py tests/integration/awslambda/test_lambda_api.py tests/integration/awslambda/test_lambda_common.py tests/integration/awslambda/test_lambda_integration_sqs.py tests/integration/cloudformation/resources/test_lambda.py tests/integration/awslambda/test_lambda_integration_dynamodbstreams.py tests/integration/awslambda/test_lambda_integration_kinesis.py tests/integration/awslambda/test_lambda_developer_tools.py"
             PYTEST_ARGS: "--reruns 3 --junitxml=target/reports/lambda_asf.xml -o junit_suite_name='lambda_asf'"
             COVERAGE_ARGS: "-p"
           command: make test-coverage
diff --git a/localstack/constants.py b/localstack/constants.py
index 0a5d741664fc6..8ae4bc2a97168 100644
--- a/localstack/constants.py
+++ b/localstack/constants.py
@@ -199,7 +199,9 @@
 DEFAULT_DEVELOP_PORT = 5678
 
 # Default bucket name of the s3 bucket used for local lambda development
-DEFAULT_BUCKET_MARKER_LOCAL = "__local__"
+# This name should be accepted by all IaC tools, so should respect s3 bucket naming conventions
+DEFAULT_BUCKET_MARKER_LOCAL = "hot-reloading-bucket"
+OLD_DEFAULT_BUCKET_MARKER_LOCAL = "__local__"
 
 # user that starts the opensearch process if the current user is root
 OS_USER_OPENSEARCH = "localstack"
diff --git a/localstack/services/awslambda/invocation/docker_runtime_executor.py b/localstack/services/awslambda/invocation/docker_runtime_executor.py
index 9f9547952856c..6ad14c11cf483 100644
--- a/localstack/services/awslambda/invocation/docker_runtime_executor.py
+++ b/localstack/services/awslambda/invocation/docker_runtime_executor.py
@@ -23,7 +23,11 @@
     get_main_endpoint_from_container,
 )
 from localstack.services.awslambda.packages import awslambda_runtime_package
-from localstack.utils.container_utils.container_client import ContainerConfiguration
+from localstack.utils.container_utils.container_client import (
+    ContainerConfiguration,
+    VolumeBind,
+    VolumeMappings,
+)
 from localstack.utils.docker_utils import DOCKER_CLIENT as CONTAINER_CLIENT
 from localstack.utils.strings import truncate
 
@@ -219,6 +223,27 @@ def start(self, env_vars: dict[str, str]) -> None:
             network=network,
             entrypoint=RAPID_ENTRYPOINT,
         )
+        if self.function_version.config.package_type == PackageType.Zip:
+            if self.function_version.config.code.is_hot_reloading():
+                # this basically means hot reloading
+                container_config.env_vars["LOCALSTACK_HOT_RELOADING_ENABLED"] = "1"
+                if container_config.volumes is None:
+                    container_config.volumes = VolumeMappings()
+                container_config.volumes.append(
+                    VolumeBind(
+                        str(self.function_version.config.code.get_unzipped_code_location()),
+                        "/var/task",
+                        read_only=True,
+                    )
+                )
+            else:
+                container_config.copy_folders.append(
+                    (
+                        f"{str(self.function_version.config.code.get_unzipped_code_location())}/.",
+                        "/var/task",
+                    )
+                )
+
         lambda_hooks.start_docker_executor.run(container_config, self.function_version)
 
         if not container_config.image_name:
@@ -232,15 +257,10 @@ def start(self, env_vars: dict[str, str]) -> None:
             CONTAINER_CLIENT.copy_into_container(
                 self.id, str(get_runtime_client_path()), RAPID_ENTRYPOINT
             )
-        if self.function_version.config.package_type == PackageType.Zip:
-            if not config.LAMBDA_PREBUILD_IMAGES:
-                CONTAINER_CLIENT.copy_into_container(
-                    self.id,
-                    f"{str(self.function_version.config.code.get_unzipped_code_location())}/.",
-                    "/var/task",
-                )
-                for source, target in container_config.copy_folders:
-                    CONTAINER_CLIENT.copy_into_container(self.id, source, target)
+        if not config.LAMBDA_PREBUILD_IMAGES:
+            # copy_folders should be empty here if package type is not zip
+            for source, target in container_config.copy_folders:
+                CONTAINER_CLIENT.copy_into_container(self.id, source, target)
 
         CONTAINER_CLIENT.start_container(self.id)
         self.ip = CONTAINER_CLIENT.get_container_ipv4_for_network(
@@ -285,8 +305,6 @@ def prepare_version(cls, function_version: FunctionVersion) -> None:
         lambda_hooks.prepare_docker_executor.run(function_version)
         if function_version.config.code:
             function_version.config.code.prepare_for_execution()
-            for layer in function_version.config.layers:
-                layer.code.prepare_for_execution()
             image_name = resolver.get_image_for_runtime(function_version.config.runtime)
             if image_name not in PULLED_IMAGES:
                 CONTAINER_CLIENT.pull_image(image_name)
diff --git a/localstack/services/awslambda/invocation/lambda_models.py b/localstack/services/awslambda/invocation/lambda_models.py
index f09dc8869f65a..d6316feb692f1 100644
--- a/localstack/services/awslambda/invocation/lambda_models.py
+++ b/localstack/services/awslambda/invocation/lambda_models.py
@@ -4,6 +4,7 @@
 import shutil
 import tempfile
 import threading
+from abc import ABCMeta, abstractmethod
 from datetime import datetime
 from pathlib import Path
 from typing import IO, TYPE_CHECKING, Dict, Optional, TypedDict
@@ -93,8 +94,55 @@ class Invocation:
     invocation_type: InvocationType
 
 
+class ArchiveCode(metaclass=ABCMeta):
+    @abstractmethod
+    def generate_presigned_url(self, endpoint_url):
+        """
+        Generates a presigned url pointing to the code archive
+        """
+        pass
+
+    @abstractmethod
+    def is_hot_reloading(self):
+        """
+        Whether this code archive is for hot reloading.
+        This means it should mount the location from the host, and should instruct the runtimes to listen for changes
+
+        :return: True if this object represents hot reloading, False otherwise
+        """
+        pass
+
+    @abstractmethod
+    def get_unzipped_code_location(self):
+        """
+        Get the location of the unzipped archive on disk
+        """
+        pass
+
+    @abstractmethod
+    def prepare_for_execution(self):
+        """
+        Unzips the code archive to the proper destination on disk, if not already present
+        """
+        pass
+
+    @abstractmethod
+    def destroy_cached(self):
+        """
+        Destroys the code object on disk, if it was saved on disk before
+        """
+        pass
+
+    @abstractmethod
+    def destroy(self):
+        """
+        Deletes the code object from S3 and the unzipped version from disk
+        """
+        pass
+
+
 @dataclasses.dataclass(frozen=True)
-class S3Code:
+class S3Code(ArchiveCode):
     """
     Objects representing a code archive stored in an internal S3 bucket.
 
@@ -149,6 +197,14 @@ def generate_presigned_url(self, endpoint_url: str | None = None) -> str:
             params["VersionId"] = self.s3_object_version
         return s3_client.generate_presigned_url("get_object", Params=params)
 
+    def is_hot_reloading(self) -> bool:
+        """
+        Whether this code archive is hot reloading
+
+        :return: True if it must it represents hot reloading, False otherwise
+        """
+        return False
+
     def get_unzipped_code_location(self) -> Path:
         """
         Get the location of the unzipped archive on disk
@@ -203,6 +259,47 @@ def destroy(self) -> None:
             )
 
 
+@dataclasses.dataclass(frozen=True)
+class HotReloadingCode(ArchiveCode):
+    """
+    Objects representing code which is mounted from a given directory from the host, for hot reloading
+    """
+
+    host_path: str
+    code_sha256: str = "hot-reloading-hash-not-available"
+    code_size: int = 0
+
+    def generate_presigned_url(self, endpoint_url: str | None = None) -> str:
+        return f"Code location: {self.host_path}"
+
+    def get_unzipped_code_location(self) -> Path:
+        return Path(self.host_path)
+
+    def is_hot_reloading(self) -> bool:
+        """
+        Whether this code archive is for hot reloading.
+        This means it should mount the location from the host, and should instruct the runtimes to listen for changes
+
+        :return: True if it represents hot reloading, False otherwise
+        """
+        return True
+
+    def prepare_for_execution(self) -> None:
+        pass
+
+    def destroy_cached(self) -> None:
+        """
+        Destroys the code object on disk, if it was saved on disk before
+        """
+        pass
+
+    def destroy(self) -> None:
+        """
+        Deletes the code object from S3 and the unzipped version from disk
+        """
+        pass
+
+
 @dataclasses.dataclass(frozen=True)
 class ImageCode:
     image_uri: str
@@ -460,7 +557,7 @@ class LayerVersion:
     layer_arn: str
 
     version: int
-    code: S3Code
+    code: ArchiveCode
     license_info: str
     compatible_runtimes: list[Runtime]
     compatible_architectures: list[Architecture]
@@ -497,7 +594,7 @@ class VersionFunctionConfiguration:
     ephemeral_storage: LambdaEphemeralStorage
 
     tracing_config_mode: TracingMode
-    code: S3Code
+    code: ArchiveCode
     last_modified: str  # ISO string
     state: VersionState
 
diff --git a/localstack/services/awslambda/invocation/lambda_service.py b/localstack/services/awslambda/invocation/lambda_service.py
index d8009f8062143..02f4a5e3993de 100644
--- a/localstack/services/awslambda/invocation/lambda_service.py
+++ b/localstack/services/awslambda/invocation/lambda_service.py
@@ -7,9 +7,11 @@
 import uuid
 from concurrent.futures import Executor, Future, ThreadPoolExecutor
 from hashlib import sha256
+from pathlib import PurePosixPath, PureWindowsPath
 from threading import RLock
 from typing import TYPE_CHECKING, Dict, Optional
 
+from localstack import config
 from localstack.aws.api.lambda_ import (
     InvalidParameterValueException,
     InvocationType,
@@ -25,8 +27,10 @@
 )
 from localstack.services.awslambda.invocation.lambda_models import (
     LAMBDA_LIMITS_CODE_SIZE_UNZIPPED_DEFAULT,
+    ArchiveCode,
     Function,
     FunctionVersion,
+    HotReloadingCode,
     ImageCode,
     Invocation,
     InvocationResult,
@@ -378,6 +382,15 @@ def store_lambda_archive(
     )
 
 
+def create_hot_reloading_code(path: str) -> HotReloadingCode:
+    # TODO extract into other function
+    if not PurePosixPath(path).is_absolute() and not PureWindowsPath(path).is_absolute():
+        raise InvalidParameterValueException(
+            f"When using hot reloading, the archive key has to be an absolute path! Your archive key: {path}",
+        )
+    return HotReloadingCode(host_path=path)
+
+
 def store_s3_bucket_archive(
     archive_bucket: str,
     archive_key: str,
@@ -385,7 +398,7 @@ def store_s3_bucket_archive(
     function_name: str,
     region_name: str,
     account_id: str,
-) -> S3Code:
+) -> ArchiveCode:
     """
     Takes the lambda archive stored in the given bucket and stores it in an internal s3 bucket
 
@@ -397,6 +410,8 @@ def store_s3_bucket_archive(
     :param account_id: account id the archive should be stored for
     :return: S3 Code object representing the archive stored in S3
     """
+    if archive_bucket == config.BUCKET_MARKER_LOCAL:
+        return create_hot_reloading_code(path=archive_key)
     s3_client: "S3Client" = aws_stack.connect_to_service("s3")
     kwargs = {"VersionId": archive_version} if archive_version else {}
     archive_file = s3_client.get_object(Bucket=archive_bucket, Key=archive_key, **kwargs)[
diff --git a/localstack/services/awslambda/lambda_api.py b/localstack/services/awslambda/lambda_api.py
index 2449d45f77adf..9dcc2a06979e8 100644
--- a/localstack/services/awslambda/lambda_api.py
+++ b/localstack/services/awslambda/lambda_api.py
@@ -22,7 +22,7 @@
 
 from flask import Flask, Response, jsonify, request
 
-from localstack import config
+from localstack import config, constants
 from localstack.aws.accounts import get_aws_account_id
 from localstack.constants import APPLICATION_JSON, LOCALHOST_HOSTNAME
 from localstack.http import Request
@@ -241,6 +241,21 @@ def build_mapping_obj(data) -> Dict:
     return mapping
 
 
+def is_hot_reloading(code: dict) -> bool:
+    bucket_name = code.get("S3Bucket")
+    if (
+        bucket_name == constants.OLD_DEFAULT_BUCKET_MARKER_LOCAL
+        and bucket_name != config.BUCKET_MARKER_LOCAL
+    ):
+        LOG.warning(
+            "Please note that using %s as local bucket marker is deprecated. Please use %s or set the config option 'BUCKET_MARKER_LOCAL'",
+            constants.OLD_DEFAULT_BUCKET_MARKER_LOCAL,
+            constants.DEFAULT_BUCKET_MARKER_LOCAL,
+        )
+        return True
+    return code.get("S3Bucket") == config.BUCKET_MARKER_LOCAL
+
+
 def format_timestamp(timestamp=None):
     timestamp = timestamp or datetime.utcnow()
     return isoformat_milliseconds(timestamp) + "+0000"
@@ -579,11 +594,11 @@ def set_archive_code(
     # get metadata
     lambda_arn = func_arn(lambda_name_or_arn)
     lambda_details = store.lambdas[lambda_arn]
-    is_local_mount = code.get("S3Bucket") == config.BUCKET_MARKER_LOCAL
+    is_local_mount = is_hot_reloading(code)
 
     if is_local_mount and config.LAMBDA_REMOTE_DOCKER:
-        msg = 'Please note that Lambda mounts (bucket name "%s") cannot be used with LAMBDA_REMOTE_DOCKER=1'
-        raise Exception(msg % config.BUCKET_MARKER_LOCAL)
+        msg = "Please note that Lambda mounts cannot be used with LAMBDA_REMOTE_DOCKER=1"
+        raise Exception(msg)
 
     # Stop/remove any containers that this arn uses.
     LAMBDA_EXECUTOR.cleanup(lambda_arn)
@@ -649,7 +664,7 @@ def store_and_get_lambda_code_archive(
     and return the Lambda CWD, file name, and zip bytes content. May optionally return None
     in case this is a Lambda with the special bucket marker __local__, used for code mounting."""
     code_passed = lambda_function.code
-    is_local_mount = code_passed.get("S3Bucket") == config.BUCKET_MARKER_LOCAL
+    is_local_mount = is_hot_reloading(code_passed)
     lambda_zip_dir = lambda_function.zip_dir
 
     if code_passed:
@@ -697,7 +712,7 @@ def generic_handler(*_):
     lambda_environment = lambda_function.envvars
     handler_name = lambda_function.handler = lambda_function.handler or LAMBDA_DEFAULT_HANDLER
     code_passed = lambda_function.code
-    is_local_mount = code_passed.get("S3Bucket") == config.BUCKET_MARKER_LOCAL
+    is_local_mount = is_hot_reloading(code_passed)
 
     # cleanup any left-over Lambda executor instances
     LAMBDA_EXECUTOR.cleanup(arn)
diff --git a/localstack/services/awslambda/packages.py b/localstack/services/awslambda/packages.py
index 8d4bb88049183..81acb1d63a8c8 100644
--- a/localstack/services/awslambda/packages.py
+++ b/localstack/services/awslambda/packages.py
@@ -9,7 +9,7 @@
 
 LAMBDA_RUNTIME_INIT_URL = "https://github.com/localstack/lambda-runtime-init/releases/download/{version}/aws-lambda-rie-{arch}"
 
-LAMBDA_RUNTIME_DEFAULT_VERSION = "v0.1.6-pre"
+LAMBDA_RUNTIME_DEFAULT_VERSION = "v0.1.7-pre"
 
 # GO Lambda runtime
 GO_RUNTIME_VERSION = "0.4.0"
@@ -21,7 +21,7 @@ def __init__(self, default_version: str = LAMBDA_RUNTIME_DEFAULT_VERSION):
         super().__init__(name="AwsLambda", default_version=default_version)
 
     def get_versions(self) -> List[str]:
-        return ["v0.1.6-pre", "v0.1.5-pre", "v0.1.4-pre", "v0.1.1-pre", "v0.1-pre"]
+        return ["v0.1.7-pre", "v0.1.6-pre", "v0.1.5-pre", "v0.1.4-pre", "v0.1.1-pre", "v0.1-pre"]
 
     def _get_installer(self, version: str) -> PackageInstaller:
         return AWSLambdaRuntimePackageInstaller(name="awslambda-runtime", version=version)
diff --git a/localstack/utils/container_utils/container_client.py b/localstack/utils/container_utils/container_client.py
index e7a84a415b715..b089552db1880 100644
--- a/localstack/utils/container_utils/container_client.py
+++ b/localstack/utils/container_utils/container_client.py
@@ -284,7 +284,7 @@ class VolumeBind:
 
     host_dir: str
     container_dir: str
-    options: Optional[List[str]] = None
+    read_only: bool = False
 
     def to_str(self) -> str:
         args = []
@@ -297,8 +297,8 @@ def to_str(self) -> str:
 
         args.append(self.container_dir)
 
-        if self.options:
-            args.append(self.options)
+        if self.read_only:
+            args.append("ro")
 
         return ":".join(args)
 
@@ -689,7 +689,7 @@ def create_container(
         tty: bool = False,
         detach: bool = False,
         command: Optional[Union[List[str], str]] = None,
-        mount_volumes: Optional[List[SimpleVolumeBind]] = None,
+        mount_volumes: Optional[Union[VolumeMappings, List[SimpleVolumeBind]]] = None,
         ports: Optional[PortMappings] = None,
         env_vars: Optional[Dict[str, str]] = None,
         user: Optional[str] = None,
@@ -972,12 +972,22 @@ def parse_additional_flags(
 
     @staticmethod
     def convert_mount_list_to_dict(
-        mount_volumes: List[SimpleVolumeBind],
+        mount_volumes: Union[List[SimpleVolumeBind], VolumeMappings],
     ) -> Dict[str, Dict[str, str]]:
         """Converts a List of (host_path, container_path) tuples to a Dict suitable as volume argument for docker sdk"""
+
+        def _map_to_dict(paths: SimpleVolumeBind | VolumeBind):
+            if isinstance(paths, VolumeBind):
+                return str(paths.host_dir), {
+                    "bind": paths.container_dir,
+                    "mode": "ro" if paths.read_only else "rw",
+                }
+            else:
+                return str(paths[0]), {"bind": paths[1], "mode": "rw"}
+
         return dict(
             map(
-                lambda paths: (str(paths[0]), {"bind": paths[1], "mode": "rw"}),
+                _map_to_dict,
                 mount_volumes,
             )
         )
diff --git a/localstack/utils/container_utils/docker_cmd_client.py b/localstack/utils/container_utils/docker_cmd_client.py
index b3b1b3850cbb4..4d97825de1775 100644
--- a/localstack/utils/container_utils/docker_cmd_client.py
+++ b/localstack/utils/container_utils/docker_cmd_client.py
@@ -22,6 +22,7 @@
     RegistryConnectionError,
     SimpleVolumeBind,
     Util,
+    VolumeBind,
 )
 from localstack.utils.run import run
 from localstack.utils.strings import to_str
@@ -628,8 +629,8 @@ def _build_run_create_cmd(
         if mount_volumes:
             cmd += [
                 volume
-                for host_path, docker_path in dict(mount_volumes).items()
-                for volume in ["-v", f"{host_path}:{docker_path}"]
+                for mount_volume in mount_volumes
+                for volume in ["-v", self._map_to_volume_param(mount_volume)]
             ]
         if interactive:
             cmd.append("--interactive")
@@ -664,3 +665,20 @@ def _build_run_create_cmd(
         if command:
             cmd += command if isinstance(command, List) else [command]
         return cmd, env_file
+
+    @staticmethod
+    def _map_to_volume_param(mount_volume: Union[SimpleVolumeBind, VolumeBind]) -> str:
+        """
+        Maps the mount volume, to a parameter for the -v docker cli argument.
+
+        Examples:
+        (host_path, container_path) -> host_path:container_path
+        VolumeBind(host_dir=host_path, container_dir=container_path, read_only=True) -> host_path:container_path:ro
+
+        :param mount_volume: Either a SimpleVolumeBind, in essence a tuple (host_dir, container_dir), or a VolumeBind object
+        :return: String which is passable as parameter to the docker cli -v option
+        """
+        if isinstance(mount_volume, VolumeBind):
+            return mount_volume.to_str()
+        else:
+            return f"{mount_volume[0]}:{mount_volume[1]}"
diff --git a/tests/integration/awslambda/functions/hot-reloading/nodejs/handler.mjs b/tests/integration/awslambda/functions/hot-reloading/nodejs/handler.mjs
new file mode 100644
index 0000000000000..7f5e9b8337a56
--- /dev/null
+++ b/tests/integration/awslambda/functions/hot-reloading/nodejs/handler.mjs
@@ -0,0 +1,11 @@
+const testConstant = "value1";
+let testCounter = 0;
+
+export const handler = async(event) => {
+    testCounter++;
+
+    return {
+        counter: testCounter,
+        constant: testConstant,
+    };
+};
diff --git a/tests/integration/awslambda/functions/hot-reloading/python/handler.py b/tests/integration/awslambda/functions/hot-reloading/python/handler.py
new file mode 100644
index 0000000000000..7f7af28f5d226
--- /dev/null
+++ b/tests/integration/awslambda/functions/hot-reloading/python/handler.py
@@ -0,0 +1,8 @@
+CONSTANT_VARIABLE = "value1"
+COUNTER = 0
+
+
+def handler(event, context):
+    global COUNTER
+    COUNTER += 1
+    return {"counter": COUNTER, "constant": CONSTANT_VARIABLE}
diff --git a/tests/integration/awslambda/test_lambda_developer_tools.py b/tests/integration/awslambda/test_lambda_developer_tools.py
new file mode 100644
index 0000000000000..1f73f4d60cbb5
--- /dev/null
+++ b/tests/integration/awslambda/test_lambda_developer_tools.py
@@ -0,0 +1,99 @@
+import json
+import os
+import time
+
+import pytest
+
+from localstack import config
+from localstack.aws.api.lambda_ import Runtime
+from localstack.testing.aws.lambda_utils import is_old_provider
+from localstack.utils.docker_utils import get_host_path_for_path_in_docker
+from localstack.utils.files import load_file, mkdir, rm_rf
+from localstack.utils.strings import short_uid
+from tests.integration.awslambda.test_lambda import THIS_FOLDER
+
+HOT_RELOADING_NODEJS_HANDLER = os.path.join(
+    THIS_FOLDER, "functions/hot-reloading/nodejs/handler.mjs"
+)
+HOT_RELOADING_PYTHON_HANDLER = os.path.join(
+    THIS_FOLDER, "functions/hot-reloading/python/handler.py"
+)
+
+
+class TestHotReloading:
+    @pytest.mark.skipif(condition=is_old_provider(), reason="Focussing on the new provider")
+    @pytest.mark.parametrize(
+        "runtime,handler_file,handler_filename",
+        [
+            (Runtime.nodejs18_x, HOT_RELOADING_NODEJS_HANDLER, "handler.mjs"),
+            (Runtime.python3_9, HOT_RELOADING_PYTHON_HANDLER, "handler.py"),
+        ],
+        ids=["nodejs18.x", "python3.9"],
+    )
+    def test_hot_reloading(
+        self,
+        create_lambda_function_aws,
+        lambda_client,
+        runtime,
+        handler_file,
+        handler_filename,
+        lambda_su_role,
+        cleanups,
+    ):
+        """Test hot reloading of lambda code"""
+        function_name = f"test-hot-reloading-{short_uid()}"
+        hot_reloading_bucket = config.BUCKET_MARKER_LOCAL
+        tmp_path = config.dirs.tmp
+        hot_reloading_dir_path = os.path.join(tmp_path, f"hot-reload-{short_uid()}")
+        mkdir(hot_reloading_dir_path)
+        cleanups.append(lambda: rm_rf(hot_reloading_dir_path))
+        function_content = load_file(handler_file)
+        with open(os.path.join(hot_reloading_dir_path, handler_filename), mode="wt") as f:
+            f.write(function_content)
+
+        mount_path = get_host_path_for_path_in_docker(hot_reloading_dir_path)
+        create_lambda_function_aws(
+            FunctionName=function_name,
+            Handler="handler.handler",
+            Code={"S3Bucket": hot_reloading_bucket, "S3Key": mount_path},
+            Role=lambda_su_role,
+            Runtime=runtime,
+        )
+        response = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")
+        response_dict = json.loads(response["Payload"].read())
+        assert response_dict["counter"] == 1
+        assert response_dict["constant"] == "value1"
+        response = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")
+        response_dict = json.loads(response["Payload"].read())
+        assert response_dict["counter"] == 2
+        assert response_dict["constant"] == "value1"
+        with open(os.path.join(hot_reloading_dir_path, handler_filename), mode="wt") as f:
+            f.write(function_content.replace("value1", "value2"))
+        # we have to sleep here, since the hot reloading is debounced with 500ms
+        time.sleep(0.6)
+        response = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")
+        response_dict = json.loads(response["Payload"].read())
+        assert response_dict["counter"] == 1
+        assert response_dict["constant"] == "value2"
+        response = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")
+        response_dict = json.loads(response["Payload"].read())
+        assert response_dict["counter"] == 2
+        assert response_dict["constant"] == "value2"
+
+        # test subdirs
+        test_folder = os.path.join(hot_reloading_dir_path, "test-folder")
+        mkdir(test_folder)
+        # make sure the creation of the folder triggered reload
+        time.sleep(0.6)
+        response = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")
+        response_dict = json.loads(response["Payload"].read())
+        assert response_dict["counter"] == 1
+        assert response_dict["constant"] == "value2"
+        # now writing something in the new folder to check if it will reload
+        with open(os.path.join(test_folder, "test-file"), mode="wt") as f:
+            f.write("test-content")
+        time.sleep(0.6)
+        response = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")
+        response_dict = json.loads(response["Payload"].read())
+        assert response_dict["counter"] == 1
+        assert response_dict["constant"] == "value2"


Code review comments:
diff_hunk: @@ -579,11 +594,11 @@ def set_archive_code(
     # get metadata
     lambda_arn = func_arn(lambda_name_or_arn)
     lambda_details = store.lambdas[lambda_arn]
-    is_local_mount = code.get("S3Bucket") == config.BUCKET_MARKER_LOCAL
+    is_local_mount = is_hot_reloading(code)
 
     if is_local_mount and config.LAMBDA_REMOTE_DOCKER:
-        msg = 'Please note that Lambda mounts (bucket name "%s") cannot be used with LAMBDA_REMOTE_DOCKER=1'
-        raise Exception(msg % config.BUCKET_MARKER_LOCAL)
+        msg = "Please note that Lambda mounts cannot be used with LAMBDA_REMOTE_DOCKER=1"
+        raise Exception(msg)
path: localstack/services/awslambda/lambda_api.py
text: @user1:
```suggestion
        raise Exception("Please note that Lambda mounts cannot be used with LAMBDA_REMOTE_DOCKER=1")
```

diff_hunk: @@ -199,7 +199,9 @@
 DEFAULT_DEVELOP_PORT = 5678
 
 # Default bucket name of the s3 bucket used for local lambda development
-DEFAULT_BUCKET_MARKER_LOCAL = "__local__"
+# This name should be accepted by all IaC tools, so should respect s3 bucket naming conventions
+DEFAULT_BUCKET_MARKER_LOCAL = "hot-reloading-bucket"
+OLD_DEFAULT_BUCKET_MARKER_LOCAL = "__local__"
path: localstack/constants.py
text: @user1:
we've started using `LEGACY_` as prefix for older variables. should we maybe do that here as well?

@author:
Already changed :)

diff_hunk: @@ -199,7 +199,9 @@
 DEFAULT_DEVELOP_PORT = 5678
 
 # Default bucket name of the s3 bucket used for local lambda development
-DEFAULT_BUCKET_MARKER_LOCAL = "__local__"
+# This name should be accepted by all IaC tools, so should respect s3 bucket naming conventions
+DEFAULT_BUCKET_MARKER_LOCAL = "hot-reloading-bucket"
+OLD_DEFAULT_BUCKET_MARKER_LOCAL = "__local__"
path: localstack/constants.py
text: @user1:
```suggestion
LEGACY_DEFAULT_BUCKET_MARKER_LOCAL = "__local__"
```

Would be a bit more inline with the conventions in config.py for example

diff_hunk: @@ -219,6 +223,27 @@ def start(self, env_vars: dict[str, str]) -> None:
             network=network,
             entrypoint=RAPID_ENTRYPOINT,
         )
+        if self.function_version.config.package_type == PackageType.Zip:
+            if self.function_version.config.code.is_hot_reloading():
+                # this basically means hot reloading
path: localstack/services/awslambda/invocation/docker_runtime_executor.py
text: @user1:

this seems a bit redundant in the `is_hot_reloading()` branch 

@author:
That method was called differently before, I renamed it to make it more clear, and the comment was left, good catch!