Various fixes to make run with sample & cleanup

Author: Sushisource

temporalio/contrib/aws/lambda_worker/__init__.py

@@ -1,8 +1,8 @@
 """A wrapper for running Temporal workers inside AWS Lambda.
 
-A single :py:func:`run_worker` call handles the full per-invocation lifecycle:
-connecting to the Temporal server, creating a worker with Lambda-tuned defaults,
-polling for tasks, and gracefully shutting down before the invocation deadline.
+A single :py:func:`run_worker` call handles the full per-invocation lifecycle: connecting to the
+Temporal server, creating a worker with Lambda-tuned defaults, polling for tasks, and gracefully
+shutting down before the invocation deadline.
 
 Quick start::
 
@@ -14,7 +14,7 @@ def configure(config: LambdaWorkerConfig) -> None:
         config.worker_config["workflows"] = [MyWorkflow]
         config.worker_config["activities"] = [my_activity]
 
-    handler = run_worker(
+    lambda_handler = run_worker(
         WorkerDeploymentVersion(
             deployment_name="my-service",
             build_id="v1.0",
@@ -24,20 +24,20 @@ def configure(config: LambdaWorkerConfig) -> None:
 
 Configuration
 -------------
-Client connection settings (address, namespace, TLS, API key) are loaded
-automatically from a TOML config file and/or environment variables via
-:py:mod:`temporalio.envconfig`. The config file is resolved in order:
+Client connection settings (address, namespace, TLS, API key) are loaded automatically from a TOML
+config file and/or environment variables via :py:mod:`temporalio.envconfig`. The config file is
+resolved in order:
 
 1. ``TEMPORAL_CONFIG_FILE`` env var, if set.
 2. ``temporal.toml`` in ``$LAMBDA_TASK_ROOT`` (typically ``/var/task``).
 3. ``temporal.toml`` in the current working directory.
 
 The file is optional -- if absent, only environment variables are used.
 
-The configure callback receives a :py:class:`LambdaWorkerConfig` dataclass with
-fields pre-populated with Lambda-appropriate defaults. Override any field
-directly in the callback. The ``task_queue`` key in ``worker_config`` is
-pre-populated from the ``TEMPORAL_TASK_QUEUE`` environment variable if set.
+The configure callback receives a :py:class:`LambdaWorkerConfig` dataclass with fields pre-populated
+with Lambda-appropriate defaults. Override any field directly in the callback. The ``task_queue``
+key in ``worker_config`` is pre-populated from the ``TEMPORAL_TASK_QUEUE`` environment variable if
+set.
 """
 
 from temporalio.contrib.aws.lambda_worker._configure import (

temporalio/contrib/aws/lambda_worker/_configure.py

@@ -17,8 +17,8 @@
 class LambdaClientConnectConfig(ClientConnectConfig, total=False):
     """Keyword arguments for :py:meth:`temporalio.client.Client.connect`.
 
-    Extends :py:class:`~temporalio.envconfig.ClientConnectConfig` with
-    additional keys that may be set by the Lambda worker or its OTel helpers.
+    Extends :py:class:`~temporalio.envconfig.ClientConnectConfig` with additional keys that may be
+    set by the Lambda worker or its OTel helpers.
     """
 
     identity: str
@@ -31,13 +31,12 @@ class LambdaClientConnectConfig(ClientConnectConfig, total=False):
 class LambdaWorkerConfig:
     """Passed to the configure callback of :py:func:`run_worker`.
 
-    Fields are pre-populated with Lambda-appropriate defaults before the
-    configure callback is invoked; the callback may read and override any of
-    them.
+    Fields are pre-populated with Lambda-appropriate defaults before the configure callback is
+    invoked; the callback may read and override any of them.
 
-    Use ``worker_config`` to set task queue, register workflows/activities, and
-    tune worker options. The ``task_queue`` key is pre-populated from the
-    ``TEMPORAL_TASK_QUEUE`` environment variable if set.
+    Use ``worker_config`` to set task queue, register workflows/activities, and tune worker options.
+    The ``task_queue`` key is pre-populated from the ``TEMPORAL_TASK_QUEUE`` environment variable if
+    set.
 
     Attributes:
         client_connect_config: Keyword arguments that will be passed to

temporalio/contrib/aws/lambda_worker/_defaults.py

@@ -18,7 +18,7 @@
 DEFAULT_MAX_CONCURRENT_NEXUS_TASKS: int = 5
 DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT: timedelta = timedelta(seconds=5)
 DEFAULT_SHUTDOWN_HOOK_BUFFER: timedelta = timedelta(seconds=2)
-DEFAULT_MAX_CACHED_WORKFLOWS: int = 100
+DEFAULT_MAX_CACHED_WORKFLOWS: int = 30
 
 DEFAULT_WORKFLOW_TASK_POLLER_BEHAVIOR = PollerBehaviorSimpleMaximum(maximum=2)
 DEFAULT_ACTIVITY_TASK_POLLER_BEHAVIOR = PollerBehaviorSimpleMaximum(maximum=1)
@@ -68,7 +68,7 @@ def build_lambda_identity(request_id: str, function_arn: str) -> str:
 
 def lambda_default_config_file_path(
     getenv: Callable[[str], str] = os.environ.get,  # type: ignore[assignment]
-) -> str:
+) -> Path:
     """Return the config file path for a Lambda environment.
 
     Resolution order:
@@ -79,6 +79,6 @@ def lambda_default_config_file_path(
     """
     config_file = getenv(ENV_CONFIG_FILE)
     if config_file:
-        return config_file
+        return Path(config_file)
     root = getenv(ENV_LAMBDA_TASK_ROOT) or "."
-    return str(Path(root) / DEFAULT_CONFIG_FILE)
+    return Path(root) / DEFAULT_CONFIG_FILE

temporalio/contrib/aws/lambda_worker/_run_worker.py

@@ -1,5 +1,3 @@
-"""Core run_worker implementation for Lambda."""
-
 from __future__ import annotations
 
 import asyncio
@@ -67,23 +65,21 @@ def _default_extract_lambda_ctx(
 def run_worker(
     version: WorkerDeploymentVersion,
     configure: Callable[[LambdaWorkerConfig], None],
-) -> Callable[[Any, Any], Awaitable[None]]:
+) -> Callable[[Any, Any], None]:
     """Create a Temporal worker Lambda handler.
 
-    Calls the *configure* callback to collect workflow/activity registrations
-    and option overrides, then returns an async Lambda handler function. On
-    each invocation the handler connects to the Temporal server, starts a
-    worker with Lambda-tuned defaults, polls for tasks until the invocation
+    Calls the *configure* callback to collect workflow/activity registrations and option overrides,
+    then returns a Lambda handler function. On each invocation the handler connects to the Temporal
+    server, starts a worker with Lambda-tuned defaults, polls for tasks until the invocation
     deadline approaches, and then gracefully shuts down.
 
-    The *version* parameter identifies this worker's deployment version.
-    ``run_worker`` always enables Worker Deployment Versioning
-    (``use_worker_versioning=True``). To provide a default versioning behavior
-    for workflows that do not specify one at registration time, set
+    The *version* parameter identifies this worker's deployment version. ``run_worker`` always
+    enables Worker Deployment Versioning (``use_worker_versioning=True``). To provide a default
+    versioning behavior for workflows that do not specify one at registration time, set
     ``deployment_config`` in ``worker_config`` in the configure callback.
 
-    The returned handler has the signature ``async handler(event, context)``
-    and should be set as your Lambda function's handler entry point.
+    The returned handler has the signature ``handler(event, context)`` and should be set as your
+    Lambda function's handler entry point.
 
     Args:
         version: The worker deployment version. Required.
@@ -92,7 +88,7 @@ def run_worker(
             activities, and options on it.
 
     Returns:
-        An async Lambda handler function.
+        A Lambda handler function.
 
     Example::
 
@@ -127,7 +123,7 @@ def _run_worker_internal(
     version: WorkerDeploymentVersion,
     configure: Callable[[LambdaWorkerConfig], None],
     deps: _WorkerDeps,
-) -> Callable[[Any, Any], Awaitable[None]]:
+) -> Callable[[Any, Any], None]:
     """Core logic with injected dependencies for testability."""
     if not version.deployment_name or not version.build_id:
         raise ValueError(
@@ -180,12 +176,14 @@ def _run_worker_internal(
 
     extract_lambda_ctx = deps.extract_lambda_ctx or _default_extract_lambda_ctx
 
-    async def _handler(_event: Any, lambda_context: Any) -> None:
-        await _invocation_handler(
-            lambda_context=lambda_context,
-            config=config,
-            deps=deps,
-            extract_lambda_ctx=extract_lambda_ctx,
+    def _handler(_event: Any, lambda_context: Any) -> None:
+        asyncio.run(
+            _invocation_handler(
+                lambda_context=lambda_context,
+                config=config,
+                deps=deps,
+                extract_lambda_ctx=extract_lambda_ctx,
+            )
         )
 
     return _handler
@@ -210,9 +208,10 @@ async def _invocation_handler(
         work_time = remaining - shutdown_buffer
         if work_time <= timedelta(seconds=1):
             raise RuntimeError(
-                f"Lambda timeout leaves almost no time for work "
-                f"(work_time={work_time}, shutdown_buffer={shutdown_buffer}); "
-                f"increase the function timeout or decrease the shutdown "
+                f"Lambda timeout is too short: {remaining.total_seconds():.1f}s "
+                f"remaining but {shutdown_buffer.total_seconds():.1f}s is "
+                f"reserved for shutdown, leaving no time for work. "
+                f"Increase the function timeout or decrease the shutdown "
                 f"deadline buffer"
             )
         elif work_time < timedelta(seconds=5):

temporalio/contrib/aws/lambda_worker/otel.py

@@ -1,12 +1,11 @@
 """OpenTelemetry helpers for Temporal workers running inside AWS Lambda.
 
-Use :py:func:`apply_defaults` inside a :py:func:`~lambda_worker.run_worker`
-configure callback for a batteries-included setup that creates an OTel
-collector exporter and tracing interceptor, suitable for use with the AWS
-Distro for OpenTelemetry (ADOT) Lambda layer.
+Use :py:func:`apply_defaults` inside a :py:func:`~lambda_worker.run_worker` configure callback for a
+batteries-included setup that creates an OTel collector exporter and tracing interceptor, suitable
+for use with the AWS Distro for OpenTelemetry (ADOT) Lambda layer.
 
-Use :py:func:`apply_tracing` or :py:func:`build_metrics_telemetry_config`
-individually if you only need one.
+Use :py:func:`apply_tracing` or :py:func:`build_metrics_telemetry_config` individually if you only
+need one.
 """
 
 from __future__ import annotations
@@ -73,23 +72,19 @@ def apply_defaults(
 ) -> None:
     """Configure OTel metrics and tracing with AWS Lambda defaults.
 
-    Sets up Core SDK metrics export via a :py:class:`~temporalio.runtime.Runtime`
-    with an :py:class:`~temporalio.runtime.OpenTelemetryConfig` pointing at the
-    OTLP collector, and adds a
-    :py:class:`~temporalio.contrib.opentelemetry.TracingInterceptor` for
-    distributed tracing.
+    Sets up Core SDK metrics export via a :py:class:`~temporalio.runtime.Runtime` with an
+    :py:class:`~temporalio.runtime.OpenTelemetryConfig` pointing at the OTLP collector, and adds a
+    :py:class:`~temporalio.contrib.opentelemetry.TracingInterceptor` for distributed tracing.
 
-    The collector endpoint defaults to ``http://localhost:4317``, which is the
-    endpoint expected by the ADOT collector Lambda layer.
+    The collector endpoint defaults to ``http://localhost:4317``, which is the endpoint expected by
+    the ADOT collector Lambda layer.
 
-    Registers a per-invocation ``ForceFlush`` shutdown hook for the
-    ``TracerProvider`` so pending traces are exported before each Lambda
-    invocation completes.
+    Registers a per-invocation ``ForceFlush`` shutdown hook for the ``TracerProvider`` so pending
+    traces are exported before each Lambda invocation completes.
 
-    Core SDK metrics are exported on the ``metric_periodicity`` interval by
-    the runtime's internal thread. There is no explicit flush API for Core
-    metrics; set ``metric_periodicity`` short enough to ensure at least one
-    export per invocation.
+    Metrics are exported on the ``metric_periodicity`` interval by the runtime's internal thread.
+    There is no explicit flush API for these metrics; set ``metric_periodicity`` short enough to
+    ensure at least one export per invocation.
 
     Args:
         config: The :py:class:`LambdaWorkerConfig` to configure.
@@ -146,16 +141,14 @@ def build_metrics_telemetry_config(
 ) -> TelemetryConfig:
     """Build a :py:class:`~temporalio.runtime.TelemetryConfig` for OTel metrics.
 
-    Returns a ``TelemetryConfig`` with
-    :py:class:`~temporalio.runtime.OpenTelemetryConfig` metrics pointed at the
-    given OTLP collector endpoint. Use this when you need to compose metrics
-    config with other telemetry settings (e.g. custom logging) into your own
+    Returns a ``TelemetryConfig`` with :py:class:`~temporalio.runtime.OpenTelemetryConfig` metrics
+    pointed at the given OTLP collector endpoint. Use this when you need to compose metrics config
+    with other telemetry settings (e.g. custom logging) into your own
     :py:class:`~temporalio.runtime.Runtime`.
 
-    Core SDK metrics are exported on the ``metric_periodicity`` interval by
-    the runtime's internal thread. There is no explicit flush API; set
-    ``metric_periodicity`` short enough to ensure at least one export per
-    Lambda invocation.
+    Core SDK metrics are exported on the ``metric_periodicity`` interval by the runtime's internal
+    thread. There is no explicit flush API; set ``metric_periodicity`` short enough to ensure at
+    least one export per Lambda invocation.
 
     Example::
 
@@ -205,8 +198,8 @@ def apply_tracing(
     """Configure only OTel tracing (no metrics) on the Lambda worker config.
 
     Adds a :py:class:`~temporalio.contrib.opentelemetry.TracingInterceptor` to
-    ``config.client_connect_config["interceptors"]`` and registers a
-    ``ForceFlush`` shutdown hook for the provider.
+    ``config.client_connect_config["interceptors"]`` and registers a ``ForceFlush`` shutdown hook
+    for the provider.
 
     Args:
         config: The :py:class:`LambdaWorkerConfig` to configure.