perf(layer): precompile Python sources in Lambda layers

.github/workflows/quality_check.yml

@@ -19,6 +19,8 @@ on:
     paths:
       - "aws_lambda_powertools/**"
       - "tests/**"
+      - "layer_v3/docker/**"
+      - "layer_v3/layer/canary/**"
       - "examples/**"
       - "pyproject.toml"
       - "uv.lock"
@@ -37,6 +39,8 @@ on:
     paths:
       - "aws_lambda_powertools/**"
       - "tests/**"
+      - "layer_v3/docker/**"
+      - "layer_v3/layer/canary/**"
       - "examples/**"
       - "pyproject.toml"
       - "uv.lock"

.github/workflows/quality_code_cdk_constructor.yml

@@ -16,6 +16,7 @@ on:
   pull_request:
     paths:
       - "layer_v3/layer_constructors/**"
+      - "layer_v3/docker/**"
       - "layer_v3/pyproject.toml"
       - "layer_v3/uv.lock"
       - ".github/workflows/quality_code_cdk_constructor.yml"
@@ -24,6 +25,7 @@ on:
   push:
     paths:
         - "layer_v3/layer_constructors/**"
+        - "layer_v3/docker/**"
         - "layer_v3/pyproject.toml"
         - "layer_v3/uv.lock"
         - ".github/workflows/quality_code_cdk_constructor.yml"

layer_v3/README.md

@@ -12,13 +12,21 @@ You can pass it as a context variable when running `synth` or `deploy`,
 cdk synth --context version=3.0.0 --pythonVersion=3.12
 ```
 
+The build precompiles Python sources for the selected runtime and retains the `.py` files for source inspection and fallback.
+Bytecode uses checked hashes so ZIP timestamp changes do not invalidate the cache.
+This increases the layer size.
+
 ## Canary stack
 
 We use a canary stack to verify that the deployment is successful and we can use the layer by adding it to a newly created Lambda function.
 The canary is deployed after the layer construct. Because the layer ARN is created during the deploy we need to pass this information async via SSM parameter.
 To achieve that we use SSM parameter store to pass the layer ARN to the canary.
 The layer stack writes the layer ARN after the deployment as SSM parameter and the canary stacks reads this information and adds the layer to the function.
 
+On creation, the canary verifies parsing, validation, source inspection, and that imports use the layer's bytecode.
+Recompilation of layer sources fails the canary before it sends a version-tracking notification.
+The cache check does not prevent deletion of the custom resource during rollback.
+
 ## Version tracking
 
 AWS Lambda versions Lambda layers by incrementing a number at the end of the ARN.

layer_v3/docker/Dockerfile

@@ -43,3 +43,7 @@ RUN cd /asset/python && \
   find . -wholename "*/tests/*" -type f -delete && \
   # remove python bytecode
   find . -regex '^.*\(__pycache__\|\.py[co]\)$' -delete
+
+# Hash validation survives ZIP timestamp changes; retain sources for inspection and fallback.
+RUN python -m compileall -q -j 2 --invalidation-mode checked-hash \
+  -s /asset -p /opt /asset/python

layer_v3/layer/canary/app.py

@@ -1,55 +1,64 @@
 import datetime
+import inspect
 import json
 import os
 import platform
 from importlib.metadata import version
 
-import boto3
-from pydantic import HttpUrl
+from bytecode_monitor import track_layer_compilation, verify_layer_bytecode
+
+# Defer cache failures to Create so CloudFormation can still delete the resource.
+with track_layer_compilation() as import_compilations:
+    import boto3
+    from pydantic import HttpUrl
+
+    from aws_lambda_powertools import Logger, Metrics, Tracer
+    from aws_lambda_powertools.event_handler import APIGatewayRestResolver
+    from aws_lambda_powertools.utilities.data_masking import DataMasking
+    from aws_lambda_powertools.utilities.parser import BaseModel, envelopes, event_parser
+    from aws_lambda_powertools.utilities.typing import LambdaContext
+    from aws_lambda_powertools.utilities.validation import validator
+
+    logger = Logger(service="version-track")
+    tracer = Tracer()  # this checks for aws-xray-sdk presence
+    metrics = Metrics(namespace="powertools-layer-canary", service="PowertoolsLayerCanary")
+    data_masker = DataMasking()
+    app = APIGatewayRestResolver()
+
+    # Model to check parser imports correctly, tests for pydantic
+    class OrderItem(BaseModel):
+        order_id: int
+        quantity: int
+        description: str
+        url: HttpUrl
+
+    # Tests for jmespath presence
+    @event_parser(model=OrderItem, envelope=envelopes.EventBridgeEnvelope)
+    def envelope_handler(event: OrderItem, context: LambdaContext):
+        return event
+
+    # Tests for fastjsonschema presence
+    @validator(
+        inbound_schema={"type": "object", "required": ["order_id", "quantity", "description", "url"]},
+        envelope="detail",
+    )
+    def validator_handler(event, context: LambdaContext):
+        pass
 
-from aws_lambda_powertools import Logger, Metrics, Tracer
-from aws_lambda_powertools.event_handler import APIGatewayRestResolver
-from aws_lambda_powertools.utilities.data_masking import DataMasking
-from aws_lambda_powertools.utilities.parser import BaseModel, envelopes, event_parser
-from aws_lambda_powertools.utilities.typing import LambdaContext
-from aws_lambda_powertools.utilities.validation import validator
-
-logger = Logger(service="version-track")
-tracer = Tracer()  # this checks for aws-xray-sdk presence
-metrics = Metrics(namespace="powertools-layer-canary", service="PowertoolsLayerCanary")
-data_masker = DataMasking()
-app = APIGatewayRestResolver()
 
 layer_arn = os.getenv("POWERTOOLS_LAYER_ARN")
 powertools_version = os.getenv("POWERTOOLS_VERSION")
 stage = os.getenv("LAYER_PIPELINE_STAGE")
 event_bus_arn = os.getenv("VERSION_TRACKING_EVENT_BUS_ARN")
 
 
-# Model to check parser imports correctly, tests for pydantic
-class OrderItem(BaseModel):
-    order_id: int
-    quantity: int
-    description: str
-    url: HttpUrl
-
-
-# Tests for jmespath presence
-@event_parser(model=OrderItem, envelope=envelopes.EventBridgeEnvelope)
-def envelope_handler(event: OrderItem, context: LambdaContext):
-    assert event.order_id != 1
-
-
-# Tests for fastjsonschema presence
-@validator(inbound_schema={}, envelope="detail")
-def validator_handler(event, context: LambdaContext):
-    pass
-
-
 def handler(event):
     logger.info("Running checks")
     check_envs()
     verify_powertools_version()
+    with track_layer_compilation() as check_compilations:
+        verify_layer_functionality()
+    verify_layer_bytecode(import_compilations + check_compilations)
     send_notification()
     return True
 
@@ -98,6 +107,32 @@ def verify_powertools_version() -> None:
     logger.info(f"Current Powertools version is: {current_version} [{_get_architecture()}]")
 
 
+def verify_layer_functionality() -> None:
+    event = {
+        "version": "0",
+        "id": "12345678-1234-1234-1234-123456789012",
+        "source": "powertools.layer.canary",
+        "account": "123456789012",
+        "time": "2026-01-01T00:00:00Z",
+        "region": "us-east-1",
+        "resources": [],
+        "detail-type": "Canary order",
+        "detail": {
+            "order_id": "2",
+            "quantity": 1,
+            "description": "Layer canary",
+            "url": "https://example.com",
+        },
+    }
+    order = envelope_handler(event, None)
+    if not isinstance(order, OrderItem) or order.order_id != 2:
+        raise ValueError("Layer failed to parse the canary event")
+    validator_handler(event, None)
+    if not inspect.getsource(Logger):
+        raise ValueError("Layer Python sources are unavailable")
+    logger.info("Layer parsing, validation, and source inspection passed")
+
+
 def send_notification():
     """
     sends an event to version tracking event bridge

layer_v3/layer/canary/bytecode_monitor.py

@@ -0,0 +1,33 @@
+from __future__ import annotations
+
+from contextlib import contextmanager
+from importlib.machinery import SourceFileLoader
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+@contextmanager
+def track_layer_compilation(layer_root: str = "/opt/python") -> Iterator[list[str]]:
+    """Record source compilation in the layer, restoring the loader even on failure."""
+    compiled_files: list[str] = []
+    source_to_code = SourceFileLoader.source_to_code
+    prefix = f"{layer_root.rstrip('/')}/"
+
+    def track(loader, data, path, *, _optimize=-1):
+        if path.startswith(prefix):
+            compiled_files.append(path)
+        return source_to_code(loader, data, path, _optimize=_optimize)
+
+    SourceFileLoader.source_to_code = track  # type: ignore[method-assign]
+    try:
+        yield compiled_files
+    finally:
+        SourceFileLoader.source_to_code = source_to_code  # type: ignore[method-assign]
+
+
+def verify_layer_bytecode(compiled_files: list[str]) -> None:
+    if compiled_files:
+        sample = ", ".join(compiled_files[:5])
+        raise ValueError(f"Layer recompiled {len(compiled_files)} source files instead of using bytecode: {sample}")

tests/functional/layer/test_bytecode.py

@@ -0,0 +1,103 @@
+import calendar
+import importlib.machinery
+import os
+import py_compile
+import types
+import zipfile
+
+import pytest
+
+from layer_v3.layer.canary.bytecode_monitor import track_layer_compilation, verify_layer_bytecode
+
+
+def load_module(source):
+    loader = importlib.machinery.SourceFileLoader("canary_bytecode_test", str(source))
+    module = types.ModuleType(loader.name)
+    exec(loader.get_code(loader.name), module.__dict__)
+    return module
+
+
+@pytest.mark.parametrize("archive_roundtrip", [False, True])
+@pytest.mark.parametrize("mode", ["source", "timestamp", "checked_hash"])
+def test_layer_bytecode_after_packaging(tmp_path, mode, archive_roundtrip):
+    layer = tmp_path / "build"
+    layer.mkdir()
+    source = layer / "sample.py"
+    source.write_text("value = 42\n")
+    # ZIP timestamps cannot represent this odd second.
+    os.utime(source, (1_700_000_001, 1_700_000_001))
+    if mode != "source":
+        invalidation = {
+            "timestamp": py_compile.PycInvalidationMode.TIMESTAMP,
+            "checked_hash": py_compile.PycInvalidationMode.CHECKED_HASH,
+        }[mode]
+        py_compile.compile(str(source), doraise=True, invalidation_mode=invalidation)
+
+    if archive_roundtrip:
+        archive_path = tmp_path / "layer.zip"
+        with zipfile.ZipFile(archive_path, "w") as archive:
+            for path in layer.rglob("*"):
+                if path.is_file():
+                    archive.write(path, path.relative_to(layer))
+        layer = tmp_path / "extracted"
+        with zipfile.ZipFile(archive_path) as archive:
+            archive.extractall(layer)
+            for info in archive.infolist():
+                timestamp = calendar.timegm(info.date_time)
+                os.utime(layer / info.filename, (timestamp, timestamp))
+        source = layer / "sample.py"
+
+    original_loader = importlib.machinery.SourceFileLoader.source_to_code
+    with track_layer_compilation(str(layer)) as compiled:
+        module = load_module(source)
+    assert module.value == 42
+    assert source.read_text() == "value = 42\n"
+    assert importlib.machinery.SourceFileLoader.source_to_code is original_loader
+    if mode == "source" or (mode == "timestamp" and archive_roundtrip):
+        assert compiled == [str(source)]
+        with pytest.raises(ValueError, match="Layer recompiled 1 source files"):
+            verify_layer_bytecode(compiled)
+    else:
+        assert compiled == []
+        verify_layer_bytecode(compiled)
+
+
+def test_checked_hash_rejects_stale_code_with_same_size_and_timestamp(tmp_path):
+    source = tmp_path / "sample.py"
+    source.write_text("value = 1\n")
+    timestamp = source.stat().st_mtime_ns
+    py_compile.compile(str(source), doraise=True, invalidation_mode=py_compile.PycInvalidationMode.CHECKED_HASH)
+    source.write_text("value = 2\n")
+    os.utime(source, ns=(timestamp, timestamp))
+
+    with track_layer_compilation(str(tmp_path)) as compiled:
+        module = load_module(source)
+
+    assert module.value == 2
+    assert compiled == [str(source)]
+
+
+def test_unrelated_imports_are_not_layer_compilations(tmp_path):
+    layer = tmp_path / "layer"
+    layer.mkdir()
+    neighbor = tmp_path / "layer-other"
+    neighbor.mkdir()
+    source = neighbor / "sample.py"
+    source.write_text("value = 42\n")
+
+    with track_layer_compilation(str(layer)) as compiled:
+        assert load_module(source).value == 42
+
+    assert compiled == []
+
+
+def test_loader_is_restored_after_failed_import(tmp_path):
+    source = tmp_path / "invalid.py"
+    source.write_text("def invalid(\n")
+    original_loader = importlib.machinery.SourceFileLoader.source_to_code
+
+    with pytest.raises(SyntaxError), track_layer_compilation(str(tmp_path)) as compiled:
+        load_module(source)
+
+    assert compiled == [str(source)]
+    assert importlib.machinery.SourceFileLoader.source_to_code is original_loader