diff --git a/packages/aws-durable-execution-sdk-python-extras/README.md b/packages/aws-durable-execution-sdk-python-extras/README.md index 273cb3f33..1ceb47c59 100644 --- a/packages/aws-durable-execution-sdk-python-extras/README.md +++ b/packages/aws-durable-execution-sdk-python-extras/README.md @@ -95,7 +95,7 @@ Every MicroVM failure is a `MicrovmError`. Catch the base class for all of them, |---|---| | `MicrovmLaunchError` | `RunMicrovm` failed after all retries. When the last attempt timed out or lost its connection, the service may have created the MicroVM anyway, and it then runs the job. So running the job again after this error can run it twice. | | `MicrovmDeliveryError` | HTTP delivery failed after all retries, or the route answered a 4xx status such as 404, or a redirect. | -| `MicrovmNotRunningError` | A session job found its MicroVM terminating, terminated, or removed. It is a `MicrovmDeliveryError`. See [Suspending an idle MicroVM](#suspending-an-idle-microvm). | +| `MicrovmNotRunningError` | A `MicrovmDeliveryError`: the MicroVM is terminating, terminated, or removed. It is not retried. A session that checks the MicroVM state before each job reports it. Any job over HTTP also reports it when `CreateMicrovmAuthToken` reports the MicroVM removed. See [Suspending an idle MicroVM](#suspending-an-idle-microvm). | | `MicrovmJobFailedError` | The job handler in the MicroVM failed, or the MicroVM sent a result that is not valid JSON. The message includes the error type that the MicroVM reported. | | `MicrovmTimeoutError` | No result arrived within `timeout`, or no heartbeat within `heartbeat_timeout`. A MicroVM that crashed ends this way. | @@ -115,6 +115,8 @@ except MicrovmError: The class is the same on the first run and on every replay, so code that branches on it stays deterministic. A `microvm` call inside your own child context, or inside a `map` or `parallel` item, reaches the code around that context as the SDK's `ChildContextError`. Its `__cause__` is the MicroVM error, so check `isinstance(error.__cause__, MicrovmTimeoutError)` there. An invalid name or config raises the SDK's `ValidationError`, and an input that is not JSON-serializable raises `TypeError`, before any durable operation starts. `Duration` raises the same `ValidationError` for a negative value. +A failed AWS call keeps its request ID in the error message, in the form ` request ID: .`, for example `RunMicrovm request ID: .`. A support case can quote it. This applies to `RunMicrovm`, `GetMicrovm`, `ResumeMicrovm`, `CreateMicrovmAuthToken`, and `TerminateMicrovm`. The SDK checkpoints the message of a failed step, and not its `__cause__`. So the request ID is in the message, the handler sees it on the first run and on replay, and the execution history keeps it. A failed `TerminateMicrovm` call does not fail the operation. So a `TerminateMicrovm` request ID appears only in the terminate log entry and in the terminate step record. A message has no request ID when the answer carries none, for example after a connection failure. A message also has no request ID when the error is not the answer of the Lambda MicroVMs call. For example, a `client` with role-assuming credentials calls STS `AssumeRole` inside each call, and an STS error carries the request ID of STS. The operation recognizes such an error by its `operation_name`. The step raises a copy of the botocore error, of the same class, with the same `response`, and with the request ID in its message. So `default_microvm_retry_strategy` decides as before. A custom `retry_strategy` receives the error that botocore raised, without the request ID. So a strategy that matches the message, such as `create_retry_strategy(RetryStrategyConfig(retryable_errors=[...]))`, decides on the message of the service. A `ResourceNotFoundException` from `GetMicrovm`, `ResumeMicrovm`, or `CreateMicrovmAuthToken` is the exception. It means that the MicroVM no longer exists. So the operation raises `MicrovmStateError` instead, and the strategy receives that error. Its `__cause__` is the botocore error. An error that the operation creates itself, such as `MicrovmStateError` or "RunMicrovm returned no microvmId", has the request ID in its message when the strategy receives it. A custom client can raise one error object for several calls. Each step message then names only its own call, and the object keeps its message. + When the error escapes the handler, the execution's error type is the class's fully-qualified name, such as `aws_durable_execution_sdk_python_extras.microvm.errors.MicrovmJobFailedError`. The SDK uses that name to rebuild the class on replay. The JavaScript operation reports the short name, `MicrovmJobFailedError`. So a rule that matches the error type of both SDKs, such as a Step Functions `Catch` or a metric filter, must match both forms. For example, match on the suffix. ### Retries in two tiers @@ -243,7 +245,11 @@ For a session that launched with either setting, each request step calls `GetMic The worker reads its idle time once, from the run hook payload of the launch. `RunMicrovm` sets the idle policy once, at the launch. So a deploy that changes `auto_suspend_on_idle` or `idle_policy` does not change what a running MicroVM does. The resume mode says whether each job checks the MicroVM state, and resumes a suspended MicroVM, before delivery. The launch checkpoint records the resume mode in its `resumeBeforeDelivery` field. Replay follows the recorded resume mode, not the current config. So a deploy changes the resume mode only for sessions that launch after it. -A MicroVM that is terminating, terminated, or removed fails the job at once with `MicrovmNotRunningError`, a subclass of `MicrovmDeliveryError`. A retry finds the same state, so the job is not retried. A session that launched with neither setting makes no `GetMicrovm` call. A terminated MicroVM's endpoint answers 502, and the request step retries a 502 as a MicroVM that is still starting. So the job fails with a plain `MicrovmDeliveryError`, after the retries of both tiers. +A MicroVM lives at most its `timeout` plus 5 minutes, and never longer than 8 hours. What the next `vm.invoke` reports after the MicroVM ends depends on the state check: + +- For a session that launched with `auto_suspend_on_idle=True`, or with an `idle_policy`, the request step calls `GetMicrovm` before each job. A MicroVM that is terminating, terminated, or removed then fails the job at once with `MicrovmNotRunningError`, a subclass of `MicrovmDeliveryError`. A retry finds the same state, so the job is not retried. The message ends with the request ID of the call that reported the state, for example `GetMicrovm request ID: .`. +- Every session, and every `microvm` job over HTTP, calls `CreateMicrovmAuthToken` before the first request, and again after a 401 or 403. The API model lists `ResourceNotFoundException` for that call. That answer fails the job at once with `MicrovmNotRunningError`, whose message ends with `CreateMicrovmAuthToken request ID: .`. The live service's answer for a removed MicroVM is not verified. +- A session that launched with neither setting makes no `GetMicrovm` call. When `CreateMicrovmAuthToken` returns a token, the request goes to the endpoint. A terminated MicroVM's endpoint answers 502, and the request step retries a 502 as a MicroVM that is still starting. So the job fails with a plain `MicrovmDeliveryError`, after the retries of both tiers. Suspending is off by default, because it needs more permissions and changes what the MicroVM does between jobs: diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/aws_errors.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/aws_errors.py new file mode 100644 index 000000000..73b0d7d9d --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/aws_errors.py @@ -0,0 +1,200 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""The request IDs of the AWS calls that the MicroVM operations make. + +The core SDK checkpoints a step failure as its type and its message, +``str(error)``. It does not checkpoint ``__cause__``, or the ``response`` of a +botocore ``ClientError``. The handler then receives a ``StepError`` rebuilt +from that record, on the first run and on replay. The request ID of a failed +call is in ``error.response["ResponseMetadata"]["RequestId"]``. So only the +message can carry the request ID to the handler and to the execution history. +""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from typing import Literal + +from aws_durable_execution_sdk_python.retries import RetryDecision +from botocore.exceptions import ClientError + + +MicrovmApiOperation = Literal[ + "RunMicrovm", + "GetMicrovm", + "ResumeMicrovm", + "CreateMicrovmAuthToken", + "TerminateMicrovm", +] +"""The Lambda MicroVMs API operations that the MicroVM operations call.""" + + +def error_code(error: ClientError) -> str | None: + """Return the error code of a botocore ``ClientError``.""" + return error.response.get("Error", {}).get("Code") + + +def is_microvms_service_error(operation: MicrovmApiOperation, error: object) -> bool: + """Return whether ``error`` is the answer of the Lambda MicroVMs call + ``operation``. + + 1. boto3 resolves the credentials inside each call. + 2. A credential provider can call another service there, such as STS + ``AssumeRole`` for an assumed role. + 3. When that call fails, the MicroVMs call raises the ``ClientError`` of + the other service. That error carries the request ID of the other + service. + 4. botocore sets ``operation_name`` to the name of the API operation that + answered. An STS error names ``AssumeRole``, not ``RunMicrovm``. + 5. So only a ``ClientError`` whose ``operation_name`` is ``operation`` + gets the name of a Lambda MicroVMs operation next to its request ID. + """ + return isinstance(error, ClientError) and error.operation_name == operation + + +def request_id_sentence(operation: MicrovmApiOperation, response: object) -> str: + """Return `` request ID: .`` for an AWS call. + + ``response`` is the output of a boto3 call, or the ``response`` of a + ``ClientError``. Both carry the request ID in + ``ResponseMetadata.RequestId``. A custom client can return an output + without it. So the function returns ``""`` when ``response`` has no + request ID that is a non-empty string. + """ + metadata = ( + response.get("ResponseMetadata") if isinstance(response, Mapping) else None + ) + request_id = metadata.get("RequestId") if isinstance(metadata, Mapping) else None + if isinstance(request_id, str) and request_id: + return f"{operation} request ID: {request_id}." + return "" + + +def with_request_id( + message: str, operation: MicrovmApiOperation, response: object +) -> str: + """Return ``message`` followed by the sentence of + :func:`request_id_sentence`. + + The function returns ``message`` unchanged when ``response`` has no + request ID. A service message often ends without a period, as in + ``MicroVM not found: mvm-1``. So a message that does not end a sentence + gets a period before the request ID sentence. + """ + sentence = request_id_sentence(operation, response) + if not sentence: + return message + if not message: + return sentence + end = "" if message.endswith((".", "!", "?")) else "." + return f"{message}{end} {sentence}" + + +def add_request_id(operation: MicrovmApiOperation, error: ClientError) -> ClientError: + """Return a copy of ``error`` whose message ends with the request ID of + the failed call, or ``error`` itself. + + The step raises the result. The caller raises it without ``from``: the + copy already has ``error`` as its ``__cause__``, and ``error`` must not + become its own cause. + + The function makes a copy, and it does not change ``error``. The + reasons, in order: + + 1. The core SDK records ``str()`` of the raised error as the message, + and the fully-qualified name of its class as the error type. + 2. So the raised error must have the request ID in ``str()``, and the + class of ``error``. A wrapper of another class would change the + recorded error type. + 3. The copy has the class of ``error``, the same ``response``, and the + same ``operation_name``. Only ``args``, ``__cause__``, and + ``_original_error`` differ. ``args`` gives ``str()``. So + :func:`default_microvm_retry_strategy`, which reads the class, the + error code, and the HTTP status, decides the same way. + 4. A fake or a custom client can raise one error object for several + calls. ``error`` keeps its message, and each call builds its copy from + that message. So the message names only the call that raised it. + 5. A custom strategy can match the message, for example with + ``create_retry_strategy(RetryStrategyConfig(retryable_errors=[...]))``. + The longer message could change its decision. So the copy holds + ``error`` in its ``_original_error`` attribute, and + :func:`decide_on_original_error` gives the strategy ``error`` instead + of the copy. The core SDK builds the step record before it calls the + strategy. So the record keeps the message with the request ID. + + The copy is made without ``__init__``. A subclass of ``ClientError`` can + take other arguments, so ``__init__`` could fail. The copy takes + ``__dict__`` of ``error``, which holds ``response``, ``operation_name``, + and any notes. ``_original_error`` is set in the same ``__dict__``. When + ``error`` is itself a copy, its ``_original_error`` is replaced with + ``error``. So the attribute always names the error that this call + received. + + ``raise ... from ...`` sets ``__cause__``, and code can set + ``__cause__`` too. Neither changes ``_original_error``. So the strategy + receives ``error`` even after ``__cause__`` changes. No module state + holds an error. The copy holds ``error``. So ``error`` lives at least as + long as the copy, and the garbage collector frees both when nothing + else refers to them. + + The function returns ``error`` itself in these cases: + + - ``error`` is not the answer of ``operation``. See + :func:`is_microvms_service_error`. + - ``error`` has no request ID. + - The class of ``error`` builds ``str()`` from something other than + ``args``. The copy would not show the request ID. + """ + if not is_microvms_service_error(operation, error): + return error + message = with_request_id(str(error), operation, error.response) + if message == str(error): + return error + labelled = type(error).__new__(type(error)) + labelled.__dict__.update(vars(error)) + labelled.args = (message,) + if str(labelled) != message: + return error + labelled.__cause__ = error + labelled.__dict__["_original_error"] = error + return labelled + + +RetryStrategy = Callable[[Exception, int], RetryDecision] + + +def decide_on_original_error(strategy: RetryStrategy) -> RetryStrategy: + """Return a retry strategy that calls ``strategy`` with the error that + botocore raised. + + The step raises the copy that :func:`add_request_id` made. The core SDK + calls the step retry strategy with that copy. The copy holds the error + that botocore raised in its ``_original_error`` attribute. The function + passes on that error instead: the same object as before the request ID + existed, with its message. So every strategy decides as it did before. + + Any other error passes unchanged. This includes a ``ClientError`` without + ``_original_error``, and the errors that the MicroVM operation creates + itself, such as ``MicrovmStateError``. Their messages have the request + ID. + + The core SDK records the step failure before it calls the strategy. So + the record, and the handler, keep the message with the request ID. + + Neither error changes its message at any time. So a traceback that code + formats during the strategy call shows the same message as at any other + time. + """ + + def decide(error: Exception, attempt: int) -> RetryDecision: + original = ( + getattr(error, "_original_error", None) + if isinstance(error, ClientError) + else None + ) + return strategy( + original if isinstance(original, ClientError) else error, attempt + ) + + return decide diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/config.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/config.py index 27d41573a..b2eb04e8e 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/config.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/config.py @@ -138,6 +138,22 @@ class MicrovmBaseConfig: It is the second retry tier. The first tier retries inside one step attempt: the boto3 client retries each call, and the request step retries the POST. Defaults to :func:`default_microvm_retry_strategy`. + + For a failed Lambda MicroVMs call, the strategy receives the + ``ClientError`` that botocore raised, with its class, its ``response``, + and its message. The step records a copy whose message ends with the + request ID. So a strategy that matches the message, such as + ``create_retry_strategy(RetryStrategyConfig(retryable_errors=[...]))``, + decides on the message of the service. An error that the operation + creates itself, such as "RunMicrovm returned no microvmId", has the + request ID in its message when the strategy receives it. + + A ``ResourceNotFoundException`` from ``GetMicrovm``, ``ResumeMicrovm``, + or ``CreateMicrovmAuthToken`` is the exception. It means that the + MicroVM no longer exists. So the operation raises ``MicrovmStateError`` + of ``aws_durable_execution_sdk_python_extras.microvm.lifecycle`` + instead, and the strategy receives that error. Its ``__cause__`` is the + ``ClientError``, and its message ends with the request ID. """ opener: urllib.request.OpenerDirector | None = None diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/errors.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/errors.py index 2773cbf90..b5826c321 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/errors.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/errors.py @@ -91,31 +91,58 @@ class MicrovmLaunchError(MicrovmError): then runs the job, and the operation cannot terminate it, because it never learned the MicroVM's ID. The MicroVM ends at its maximum duration. So a caller that runs the job again after this error can run it twice. + + When the service answered, the message ends with + ``RunMicrovm request ID: .``. """ class MicrovmDeliveryError(MicrovmError): - """The job could not be delivered to the MicroVM.""" + """The job could not be delivered to the MicroVM. + + When a Lambda MicroVMs call failed, the message ends with its request + ID, for example ``CreateMicrovmAuthToken request ID: .``. + """ class MicrovmNotRunningError(MicrovmDeliveryError): - """The session's MicroVM can no longer run a job. - - The MicroVM is terminating, terminated, or removed. A session MicroVM - lives at most its session ``timeout`` plus 5 minutes, and never longer - than 8 hours, running or suspended. A ``TerminateMicrovm`` call from - outside the session ends it too. It is a :class:`MicrovmDeliveryError`, - so code that catches delivery failures also catches it. - - Only a session that checks the MicroVM state before each job raises it. - It checks when ``auto_suspend_on_idle`` is ``True``, or when an - ``idle_policy`` is set. A later attempt finds the same state, so the job - is not retried. To continue, start a new session. - - Without the check, the session makes no ``GetMicrovm`` call. The endpoint - of a terminated MicroVM answers 502, and the request step retries a 502 - as a MicroVM that is still starting. So the job fails with a plain - :class:`MicrovmDeliveryError`, after the retries of both tiers. + """The MicroVM can no longer run a job. + + The MicroVM is terminating, terminated, or removed. A MicroVM lives at + most its ``timeout`` plus 5 minutes, and never longer than 8 hours, + running or suspended. A ``TerminateMicrovm`` call from outside the + operation ends it too. It is a :class:`MicrovmDeliveryError`, so code + that catches delivery failures also catches it. A later attempt finds + the same state, so the job is not retried. To continue a session, start + a new session. + + Two checks raise it: + + 1. A session that checks the MicroVM state before each job. It checks + when ``auto_suspend_on_idle`` is ``True``, or when an ``idle_policy`` + is set. ``GetMicrovm`` or ``ResumeMicrovm`` then reports the state. + 2. Every job over HTTP, of a session or of :func:`microvm`. The request + step calls ``CreateMicrovmAuthToken`` before the first request, and + again after a 401 or 403. That call can answer + ``ResourceNotFoundException``. + + The message ends with the request ID of the call that reported the + state, for example ``GetMicrovm request ID: .``. The core SDK + records only the message of the failure. So a support case can quote + the request ID from the message, on the first run and on replay. + + Without the state check, a session makes no ``GetMicrovm`` call. The + request step still calls ``CreateMicrovmAuthToken`` first. When that + call answers ``ResourceNotFoundException``, the job fails with this + error. When it returns a token, the request goes to the endpoint. The + endpoint of a terminated MicroVM answers 502, and the request step + retries a 502 as a MicroVM that is still starting. So the job then fails + with a plain :class:`MicrovmDeliveryError`, after the retries of both + tiers. + + The live service's answer of ``CreateMicrovmAuthToken`` for a removed + MicroVM is not verified. The API model lists + ``ResourceNotFoundException`` for that call. """ diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/lifecycle.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/lifecycle.py index 539b5175a..5e31324ca 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/lifecycle.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/lifecycle.py @@ -8,11 +8,17 @@ import time from collections.abc import Callable from dataclasses import dataclass -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Literal from aws_durable_execution_sdk_python.types import LoggerInterface from botocore.exceptions import ClientError +from aws_durable_execution_sdk_python_extras.microvm.aws_errors import ( + add_request_id, + error_code, + is_microvms_service_error, + with_request_id, +) from aws_durable_execution_sdk_python_extras.microvm.request import ( MicrovmEndpointUnavailableError, call_before_deadline, @@ -37,8 +43,18 @@ class MicrovmStateError(Exception): """The MicroVM is terminating, terminated, or removed. The request step raises it. A later attempt finds the same state, so the - default retry strategy does not retry it. The session job then raises + default retry strategy does not retry it. The delivery then raises :class:`MicrovmNotRunningError`. + + The message ends with the request ID of the AWS call that reported the + state, when that call has one. For a MicroVM that no longer exists, + ``__cause__`` is the botocore error. Only the step retry strategy of a + live attempt sees ``__cause__``. The core SDK records the message, not + the cause. So the message carries the request ID. + + The request step records the fully-qualified name of this class as its + error type, and the operation compares the recorded type with that name. + So the class stays in this module. """ def __init__(self, message: str, state: str | None = None) -> None: @@ -79,7 +95,16 @@ def ensure_running(options: EnsureRunningOptions) -> str | None: 4. ``SUSPENDING``: it waits. The MicroVM becomes ``SUSPENDED``, and the function then resumes it. 5. ``TERMINATING``, ``TERMINATED``, or not found: it raises - :class:`MicrovmStateError`. + :class:`MicrovmStateError`. ``GetMicrovm`` or ``ResumeMicrovm`` can + report that the MicroVM is not found. The message ends with the + request ID of the call that reported the state, in the form + ``GetMicrovm request ID: .``. + + Any other ``GetMicrovm`` or ``ResumeMicrovm`` error, such as + ``AccessDeniedException``, is raised with its class and its + ``response``. Its message then ends with the request ID of the call. See + :func:`add_request_id`. A ``ConflictException`` from ``ResumeMicrovm`` + is not an error: polling continues. It polls after 250 milliseconds, and doubles the wait up to 2 seconds. It stops after ``max_wait_seconds``, or 10 seconds before the invocation @@ -96,6 +121,8 @@ def ensure_running(options: EnsureRunningOptions) -> str | None: MicrovmStateError: When the MicroVM can no longer run a job. MicrovmEndpointUnavailableError: When the MicroVM does not reach ``RUNNING`` in time. + ClientError: When ``GetMicrovm`` or ``ResumeMicrovm`` fails + otherwise. """ started = time.monotonic() reserve = reserve_starts_at(options.remaining_seconds) @@ -130,21 +157,18 @@ def resume_microvm() -> Any: unavailable, ) - def not_found() -> MicrovmStateError: - msg = ( - f"MicroVM {microvm_id} no longer exists. A session MicroVM lives at " - "most its session timeout plus 5 minutes, and never longer than 8 " - "hours." - ) - return MicrovmStateError(msg) - + # GetMicrovm and ResumeMicrovm both answer ResourceNotFoundException for + # a removed MicroVM. So both raise the same MicrovmStateError. Every other + # error of either call keeps its class and its response, and its message + # gets the request ID. add_request_id returns the error, or a copy whose + # __cause__ is the error. So that raise has no `from`. while True: try: current = get_microvm() except ClientError as error: - if _code(error) == "ResourceNotFoundException": - raise not_found() from error - raise + if error_code(error) == "ResourceNotFoundException": + raise microvm_not_found(microvm_id, "GetMicrovm", error) from error + raise add_request_id("GetMicrovm", error) state = current.get("state") last_state = state if state == "RUNNING" or (state == "PENDING" and not resume_sent): @@ -156,8 +180,10 @@ def not_found() -> MicrovmStateError: return current.get("endpoint") if state in {"TERMINATING", "TERMINATED"}: reason = current.get("stateReason") + # The GetMicrovm output carries the request ID. The state fails + # the step, so the message names that request ID. msg = f"MicroVM {microvm_id} is {state}{f': {reason}' if reason else ''}." - raise MicrovmStateError(msg, state) + raise MicrovmStateError(with_request_id(msg, "GetMicrovm", current), state) if state == "SUSPENDED" and not resume_sent: try: resume_microvm() @@ -166,11 +192,13 @@ def not_found() -> MicrovmStateError: # progress, for example an auto-resume. Polling continues # either way. ResourceNotFoundException means that the # MicroVM was removed after the state check. - code = _code(error) + code = error_code(error) if code == "ResourceNotFoundException": - raise not_found() from error + raise microvm_not_found( + microvm_id, "ResumeMicrovm", error + ) from error if code != "ConflictException": - raise + raise add_request_id("ResumeMicrovm", error) resume_sent = True options.logger.info("resuming MicroVM", extra={"state": state}) @@ -180,5 +208,31 @@ def not_found() -> MicrovmStateError: delay = min(delay * 2, _MAX_POLL_SECONDS) -def _code(error: ClientError) -> str | None: - return error.response.get("Error", {}).get("Code") +def microvm_not_found( + microvm_id: str, + operation: Literal["GetMicrovm", "ResumeMicrovm", "CreateMicrovmAuthToken"], + error: ClientError, +) -> MicrovmStateError: + """Return the :class:`MicrovmStateError` for a call that answered + ``ResourceNotFoundException``. + + ``GetMicrovm``, ``ResumeMicrovm``, and ``CreateMicrovmAuthToken`` each + answer ``ResourceNotFoundException`` for a removed MicroVM. The request + step calls all three. So each call reports a removed MicroVM with the + same error. The core SDK records only the message of a step failure. So + the message names the call and its request ID. The caller raises the + result ``from error``. The caller matches the error by its code. So the + message names a request ID only when ``error`` is the answer of + ``operation``, as in :func:`add_request_id`. + + The caller can be a session or :func:`microvm`. Both configs name the + MicroVM lifetime ``timeout``. Both launches set + ``maximumDurationInSeconds`` to ``timeout`` plus 5 minutes, at most 8 + hours. So the message states that limit in words that fit both callers. + """ + msg = ( + f"MicroVM {microvm_id} no longer exists. A MicroVM lives at most its " + "`timeout` plus 5 minutes, and never longer than 8 hours." + ) + response = error.response if is_microvms_service_error(operation, error) else None + return MicrovmStateError(with_request_id(msg, operation, response)) diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py index 2d4a926e9..60f7a78d7 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py @@ -35,6 +35,12 @@ from aws_durable_execution_sdk_python.serdes import JsonSerDes from botocore.exceptions import ClientError +from aws_durable_execution_sdk_python_extras.microvm.aws_errors import ( + add_request_id, + decide_on_original_error, + error_code, + with_request_id, +) from aws_durable_execution_sdk_python_extras.microvm.config import ( MAX_MICROVM_DURATION_SECONDS, LambdaMicrovmsClient, @@ -60,6 +66,7 @@ EnsureRunningOptions, MicrovmStateError, ensure_running, + microvm_not_found, ) from aws_durable_execution_sdk_python_extras.microvm.request import ( DEFAULT_MICROVM_JOB_PATH, @@ -251,11 +258,16 @@ def microvm( TypeError: When ``job_input`` is not JSON-serializable. No durable operation starts. MicrovmLaunchError: When ``RunMicrovm`` fails after all retries. + MicrovmNotRunningError: When ``CreateMicrovmAuthToken`` reports that + the MicroVM no longer exists. It is a ``MicrovmDeliveryError``. MicrovmDeliveryError: When the HTTP delivery fails after all retries, or the route answers a 4xx status such as 404, or a redirect. MicrovmJobFailedError: When the MicroVM reports a failure, or sends a result that is not valid JSON. MicrovmTimeoutError: When no result or heartbeat arrives in time. + + The message of a failed Lambda MicroVMs call ends with its request ID, + for example ``RunMicrovm request ID: .``. """ scope = _OperationScope.create(context, name, job_input, config) @@ -273,6 +285,22 @@ def run(child: DurableContext) -> Any: longer run a job. The SDK records the fully-qualified class name.""" +def step_retry_strategy( + config: MicrovmBaseConfig, +) -> Callable[[Exception, int], RetryDecision]: + """Return the retry strategy of the launch, request, and terminate steps. + + It is the configured strategy, or :func:`default_microvm_retry_strategy`. + It decides on the error that botocore raised, without the request ID in + the message. See :func:`decide_on_original_error`. A session passes the + same strategy to each job. So this applies to every AWS call of both + operations. + """ + return decide_on_original_error( + config.retry_strategy or default_microvm_retry_strategy + ) + + def _validate_base(name: str, config: MicrovmBaseConfig, timeout: Duration) -> int: """Validate the settings that every MicroVM operation shares. @@ -446,6 +474,11 @@ def deliver( ``GetMicrovm`` reports. A session sets it when its MicroVM can be suspended. + ``GetMicrovm``, ``ResumeMicrovm``, and ``CreateMicrovmAuthToken`` + answer ``ResourceNotFoundException`` for a removed MicroVM. An + attempt raises ``MicrovmStateError`` for that answer of any of the + three calls. The default retry strategy does not retry it. + Raises: MicrovmNotRunningError: When the MicroVM can no longer run a job. MicrovmDeliveryError: When the delivery fails after all retries. @@ -497,22 +530,42 @@ def running(max_wait_seconds: float) -> str | None: endpoint = running(retry_window_seconds) or endpoint recheck = running - send_job( - SendJobOptions( - client=client, - microvm_id=launched.microvm_id, - endpoint=endpoint, - path=path, - port=port, - body=body, - retry_window_seconds=retry_window_seconds, - remaining_seconds=remaining_seconds, - logger=step_context.logger, - opener=self.config.opener, - window_started_at=window_started_at, - recheck=recheck, + # Only CreateMicrovmAuthToken can raise ResourceNotFoundException + # here: + # + # 1. _create_auth_token raises that answer unchanged. + # 2. send_job calls new_token() outside any try. So the error + # leaves send_job unchanged. + # 3. _run_with_limit re-raises the same error object. + # 4. ensure_running turns the same answer of GetMicrovm and + # ResumeMicrovm into MicrovmStateError. So neither call raises + # a ClientError with that code. + # + # So the step raises MicrovmStateError for CreateMicrovmAuthToken. + # The default retry strategy does not retry that error. + try: + send_job( + SendJobOptions( + client=client, + microvm_id=launched.microvm_id, + endpoint=endpoint, + path=path, + port=port, + body=body, + retry_window_seconds=retry_window_seconds, + remaining_seconds=remaining_seconds, + logger=step_context.logger, + opener=self.config.opener, + window_started_at=window_started_at, + recheck=recheck, + ) ) - ) + except ClientError as error: + if error_code(error) == "ResourceNotFoundException": + raise microvm_not_found( + launched.microvm_id, "CreateMicrovmAuthToken", error + ) from error + raise try: child.step( @@ -604,15 +657,22 @@ def _launch_call( request["logging"] = dict(config.logging) if idle_policy is not None: request["idlePolicy"] = dict(idle_policy) - response = self._client().run_microvm(**request) + try: + response = self._client().run_microvm(**request) + except ClientError as error: + # add_request_id returns the error, or a copy whose __cause__ is + # the error. So the raise has no `from`. + raise add_request_id("RunMicrovm", error) + # A missing field fails the step. So the message names the request + # ID of the answer. microvm_id = response.get("microvmId") if not microvm_id: msg = "RunMicrovm returned no microvmId" - raise RuntimeError(msg) + raise RuntimeError(with_request_id(msg, "RunMicrovm", response)) endpoint = response.get("endpoint") if over_http and not endpoint: msg = "RunMicrovm returned no endpoint" - raise RuntimeError(msg) + raise RuntimeError(with_request_id(msg, "RunMicrovm", response)) return LaunchResult( microvm_id=microvm_id, delivery=delivery, @@ -626,11 +686,10 @@ def _terminate_call(self, microvm_id: str) -> None: except ClientError as error: # TerminateMicrovm succeeds for a terminated MicroVM. The service # answers ResourceNotFoundException for a MicroVM whose record it - # has removed. Either way, the MicroVM is gone. - if error.response.get("Error", {}).get("Code") != ( - "ResourceNotFoundException" - ): - raise + # has removed. Either way, the MicroVM is gone. Any other error + # gets the request ID. See add_request_id. + if error_code(error) != "ResourceNotFoundException": + raise add_request_id("TerminateMicrovm", error) def _connector_arn(self, connector: str) -> str: return ( @@ -682,7 +741,7 @@ def create( heartbeat_timeout_seconds=heartbeat_timeout_seconds, request=request, retry_window_seconds=retry_window_seconds, - retry_strategy=config.retry_strategy or default_microvm_retry_strategy, + retry_strategy=step_retry_strategy(config), ) def run_job(self, child: DurableContext, job_input: Any) -> Any: diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/request.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/request.py index c7f905c81..e562fb387 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/request.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/request.py @@ -21,6 +21,13 @@ from typing import TYPE_CHECKING, Any, TypeVar, cast from aws_durable_execution_sdk_python.types import LoggerInterface +from botocore.exceptions import ClientError + +from aws_durable_execution_sdk_python_extras.microvm.aws_errors import ( + add_request_id, + error_code, + with_request_id, +) if TYPE_CHECKING: @@ -325,7 +332,14 @@ def send_job(options: SendJobOptions) -> int: MicrovmRequestRejectedError: When the endpoint rejects the job. MicrovmRequestUnauthorizedError: When a new auth token is refused. MicrovmEndpointUnavailableError: When the tier ends. - Exception: When ``CreateMicrovmAuthToken`` fails. + ClientError: When ``CreateMicrovmAuthToken`` answers + ``ResourceNotFoundException``. It is the error that the client + raised, unchanged. The request step turns it into + ``MicrovmStateError``. + MicrovmStateError: When ``recheck`` reports that the MicroVM can no + longer run a job. + Exception: When ``CreateMicrovmAuthToken`` fails otherwise. The + message of a botocore error ends with the request ID. """ require_time_for_call(options.remaining_seconds) started = ( @@ -449,14 +463,43 @@ def new_token() -> Mapping[str, str]: def _create_auth_token(options: SendJobOptions) -> Mapping[str, str]: - response = options.client.create_microvm_auth_token( - microvmIdentifier=options.microvm_id, - expirationInMinutes=_AUTH_TOKEN_MINUTES, - allowedPorts=[{"port": options.port}], - ) + """Call ``CreateMicrovmAuthToken``, and return the token headers. + + The function handles a failed call this way: + + 1. The API model lists ``ResourceNotFoundException`` for this call. It + means that the MicroVM was removed after the launch or after the + state check. The function raises that error unchanged. The request + step in ``operation.py`` turns it into ``MicrovmStateError``, with + the error as its ``__cause__``. + 2. Any other error is raised with its class and its ``response``, and + its message ends with the request ID. See :func:`add_request_id`. + + Raises: + RuntimeError: When the answer has no ``authToken``. + ClientError: When the call fails. A ``ResourceNotFoundException`` + is the error that the client raised. Any other error can be the + copy that :func:`add_request_id` returns. + """ + try: + response = options.client.create_microvm_auth_token( + microvmIdentifier=options.microvm_id, + expirationInMinutes=_AUTH_TOKEN_MINUTES, + allowedPorts=[{"port": options.port}], + ) + except ClientError as error: + if error_code(error) == "ResourceNotFoundException": + raise + # add_request_id returns the error, or a copy whose __cause__ is + # the error. So the raise has no `from`. + raise add_request_id("CreateMicrovmAuthToken", error) token = response.get("authToken") if not token: - msg = "CreateMicrovmAuthToken returned no authToken" + msg = with_request_id( + "CreateMicrovmAuthToken returned no authToken", + "CreateMicrovmAuthToken", + response, + ) raise RuntimeError(msg) return token diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py index 3bffbfab9..1b61ef809 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py @@ -38,11 +38,9 @@ _validate_timeout, create_job_callback, job_error, + step_retry_strategy, unknown_delivery_error, ) -from aws_durable_execution_sdk_python_extras.microvm.retry import ( - default_microvm_retry_strategy, -) if TYPE_CHECKING: @@ -251,8 +249,10 @@ def invoke( context that ``microvm_session`` was called with, or ``job_input`` is not JSON-serializable. ValidationError: When ``name`` or ``options`` is not valid. - MicrovmNotRunningError: When the session's MicroVM has ended, and - the session checks the MicroVM state before each job. + MicrovmNotRunningError: When the session's MicroVM has ended. The + session reports it when it checks the MicroVM state before + each job, and for every job when ``CreateMicrovmAuthToken`` + reports the MicroVM removed. MicrovmDeliveryError: When the delivery fails after all retries, or the route rejects the job. MicrovmJobFailedError: When the MicroVM reports a failure, or @@ -370,7 +370,7 @@ def create( partition=partition, region=region, timeout_seconds=timeout_seconds, - retry_strategy=config.retry_strategy or default_microvm_retry_strategy, + retry_strategy=step_retry_strategy(config), auto_suspend_idle_seconds=auto_suspend_idle_seconds, idle_policy=config.idle_policy, outer_context=context, diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/aws_errors_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/aws_errors_test.py new file mode 100644 index 000000000..8780e67b9 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/aws_errors_test.py @@ -0,0 +1,464 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Tests for the request ID helpers of the MicroVM operations.""" + +from __future__ import annotations + +import gc +import logging +import weakref +from typing import Any + +import boto3 +import pytest +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.retries import RetryDecision +from botocore.exceptions import ClientError + +from aws_durable_execution_sdk_python_extras.microvm.aws_errors import ( + add_request_id, + decide_on_original_error, + request_id_sentence, + with_request_id, +) +from aws_durable_execution_sdk_python_extras.microvm.lifecycle import ( + MicrovmStateError, + microvm_not_found, +) +from aws_durable_execution_sdk_python_extras.microvm.request import ( + SendJobOptions, + send_job, +) + + +def response(request_id: Any = "req-1", code: str = "ThrottlingException") -> Any: + metadata: dict[str, Any] = {"HTTPStatusCode": 400} + if request_id is not None: + metadata["RequestId"] = request_id + return { + "Error": {"Code": code, "Message": "slow down"}, + "ResponseMetadata": metadata, + } + + +@pytest.mark.parametrize( + ("source", "sentence"), + [ + ({"ResponseMetadata": {"RequestId": "req-1"}}, "RunMicrovm request ID: req-1."), + ({"ResponseMetadata": {"RequestId": ""}}, ""), + ({"ResponseMetadata": {"RequestId": 7}}, ""), + ({"ResponseMetadata": {}}, ""), + ({"ResponseMetadata": "req-1"}, ""), + ({}, ""), + (None, ""), + ("req-1", ""), + ], +) +def test_names_only_a_request_id_that_is_a_non_empty_string( + source: Any, sentence: str +) -> None: + assert request_id_sentence("RunMicrovm", source) == sentence + + +@pytest.mark.parametrize( + ("message", "expected"), + [ + ("bad image", "bad image. RunMicrovm request ID: req-1."), + ("bad image.", "bad image. RunMicrovm request ID: req-1."), + ("bad image!", "bad image! RunMicrovm request ID: req-1."), + ("bad image?", "bad image? RunMicrovm request ID: req-1."), + ("", "RunMicrovm request ID: req-1."), + ], +) +def test_ends_the_message_with_the_request_id_sentence( + message: str, expected: str +) -> None: + source = {"ResponseMetadata": {"RequestId": "req-1"}} + assert with_request_id(message, "RunMicrovm", source) == expected + + +def test_keeps_the_message_without_a_request_id() -> None: + assert with_request_id("bad image", "RunMicrovm", {}) == "bad image" + + +def test_copies_a_modeled_error_with_the_request_id_in_its_message() -> None: + exceptions = boto3.client("lambda-microvms", region_name="us-west-2").exceptions + error = exceptions.ThrottlingException(response(), "RunMicrovm") + error.add_note("a note") + original = str(error) + + labelled = add_request_id("RunMicrovm", error) + + assert labelled is not error + assert type(labelled) is type(error) + assert labelled.response is error.response + assert labelled.operation_name == "RunMicrovm" + assert labelled.__notes__ == ["a note"] + assert labelled.__cause__ is error + assert vars(labelled)["_original_error"] is error + assert "_original_error" not in vars(error) + assert labelled.__suppress_context__ + assert str(labelled) == f"{original}. RunMicrovm request ID: req-1." + assert labelled.args == (str(labelled),) + # The error that botocore raised keeps its message. + assert str(error) == original + + +@pytest.mark.parametrize( + ("error", "operation"), + [ + # No request ID. + (ClientError(response(None), "RunMicrovm"), "RunMicrovm"), + (ClientError(response(""), "RunMicrovm"), "RunMicrovm"), + # The answer of another operation, such as STS AssumeRole in a + # credential provider. + (ClientError(response(), "AssumeRole"), "RunMicrovm"), + (ClientError(response(), "CreateMicrovmAuthToken"), "TerminateMicrovm"), + ], +) +def test_returns_the_error_itself_when_it_gets_no_request_id( + error: ClientError, operation: Any +) -> None: + original = str(error) + + assert add_request_id(operation, error) is error + assert str(error) == original + assert error.__cause__ is None + + +def test_returns_the_error_itself_when_its_class_builds_str_otherwise() -> None: + class FixedMessageError(ClientError): + def __str__(self) -> str: + return "fixed" + + error = FixedMessageError(response(), "RunMicrovm") + + assert add_request_id("RunMicrovm", error) is error + + +def test_copies_a_subclass_whose_init_takes_other_arguments() -> None: + class WrappedError(ClientError): + def __init__(self, detail: str) -> None: + super().__init__(response(), "GetMicrovm") + self.detail = detail + + error = WrappedError("extra") + + labelled = add_request_id("GetMicrovm", error) + + assert type(labelled) is WrappedError + assert labelled.detail == "extra" # type: ignore[attr-defined] + assert str(labelled).endswith(". GetMicrovm request ID: req-1.") + + +def test_builds_each_copy_from_the_message_of_the_error_that_was_raised() -> None: + error = ClientError(response("req-1"), "RunMicrovm") + first = add_request_id("RunMicrovm", error) + error.response["ResponseMetadata"]["RequestId"] = "req-2" + + second = add_request_id("RunMicrovm", error) + + assert str(first).endswith(" RunMicrovm request ID: req-1.") + assert str(second).endswith(". RunMicrovm request ID: req-2.") + assert "req-1" not in str(second) + + +def test_gives_the_strategy_the_error_that_botocore_raised() -> None: + error = ClientError(response(), "RunMicrovm") + labelled = add_request_id("RunMicrovm", error) + seen: list[tuple[Exception, int]] = [] + decision = RetryDecision.retry(Duration.from_seconds(3)) + + def strategy(received: Exception, attempt: int) -> RetryDecision: + seen.append((received, attempt)) + return decision + + assert decide_on_original_error(strategy)(labelled, 2) is decision + assert seen == [(error, 2)] + assert seen[0][0] is error + + +@pytest.mark.parametrize( + "error", + [ + RuntimeError("RunMicrovm returned no microvmId. RunMicrovm request ID: r."), + MicrovmStateError("MicroVM mvm-1 is TERMINATED."), + ClientError(response(), "RunMicrovm"), + ], + ids=["builtin", "state", "client-error"], +) +def test_passes_any_other_error_unchanged(error: Exception) -> None: + seen: list[Exception] = [] + + def strategy(received: Exception, attempt: int) -> RetryDecision: + seen.append(received) + return RetryDecision.no_retry() + + decide_on_original_error(strategy)(error, 1) + + assert seen == [error] + assert seen[0] is error + + +def test_passes_a_client_error_whose_original_is_not_a_client_error() -> None: + """Only a ``ClientError`` in ``_original_error`` replaces the error.""" + error = ClientError(response(), "RunMicrovm") + error.__dict__["_original_error"] = RuntimeError("not botocore") + seen: list[Exception] = [] + + def strategy(received: Exception, attempt: int) -> RetryDecision: + seen.append(received) + return RetryDecision.no_retry() + + decide_on_original_error(strategy)(error, 1) + + assert seen[0] is error + + +def test_raises_what_the_strategy_raises() -> None: + def strategy(received: Exception, attempt: int) -> RetryDecision: + msg = "strategy bug" + raise LookupError(msg) + + labelled = add_request_id("RunMicrovm", ClientError(response(), "RunMicrovm")) + + with pytest.raises(LookupError, match="strategy bug"): + decide_on_original_error(strategy)(labelled, 1) + + +@pytest.mark.parametrize( + ("operation", "error", "sentence"), + [ + ( + "CreateMicrovmAuthToken", + ClientError( + response(code="ResourceNotFoundException"), "CreateMicrovmAuthToken" + ), + " CreateMicrovmAuthToken request ID: req-1.", + ), + ( + "GetMicrovm", + ClientError(response(None, "ResourceNotFoundException"), "GetMicrovm"), + "", + ), + # The caller matches the code. An error of another operation with the + # same code gets no request ID. + ( + "ResumeMicrovm", + ClientError(response(code="ResourceNotFoundException"), "AssumeRole"), + "", + ), + ], +) +def test_reports_a_removed_microvm_in_words_that_fit_both_operations( + operation: Any, error: ClientError, sentence: str +) -> None: + state_error = microvm_not_found("mvm-1", operation, error) + + assert state_error.state is None + assert str(state_error) == ( + "MicroVM mvm-1 no longer exists. A MicroVM lives at most its `timeout` " + "plus 5 minutes, and never longer than 8 hours." + sentence + ) + + +class _TokenClient: + """Answers CreateMicrovmAuthToken from a list.""" + + def __init__(self, answers: list[Any]) -> None: + self._answers = answers + + def create_microvm_auth_token(self, **request: Any) -> Any: + answer = self._answers.pop(0) + if isinstance(answer, BaseException): + raise answer + return answer + + +@pytest.mark.parametrize( + ("statuses", "before"), + [([], []), ([403], [{"authToken": {"x-t": "t"}}])], + ids=["first-token", "new-token-after-403"], +) +@pytest.mark.parametrize( + "remaining", [None, 600.0], ids=["no-deadline", "in-a-daemon-thread"] +) +def test_send_job_raises_the_not_found_answer_of_the_token_call_unchanged( + fake_endpoint: Any, + statuses: list[int], + before: list[Any], + remaining: float | None, +) -> None: + """The request step turns the answer into ``MicrovmStateError``. So + ``send_job`` raises the error that the client raised, without a copy. + With a deadline, the call runs in the daemon thread of + ``_run_with_limit``, which re-raises the same object.""" + endpoint = fake_endpoint(statuses) + not_found = ClientError( + response(code="ResourceNotFoundException"), "CreateMicrovmAuthToken" + ) + original = str(not_found) + options = SendJobOptions( + client=_TokenClient([*before, not_found]), # type: ignore[arg-type] + microvm_id="mvm-1", + endpoint=endpoint.url, + path="/job", + port=8080, + body=b"{}", + retry_window_seconds=60, + remaining_seconds=lambda: remaining, + logger=logging.getLogger("test"), + sleep=lambda seconds: None, + ) + + with pytest.raises(ClientError) as raised: + send_job(options) + + assert raised.value is not_found + assert str(not_found) == original + assert len(endpoint.requests) == len(statuses) + + +def _frame_names(error: BaseException) -> list[str]: + names = [] + traceback = error.__traceback__ + while traceback is not None: + names.append(traceback.tb_frame.f_code.co_name) + traceback = traceback.tb_next + return names + + +def _fail_a_token_call_under_a_deadline( + endpoint_url: str, +) -> tuple[weakref.ref[ClientError], weakref.ref[ClientError]]: + """Fail one ``CreateMicrovmAuthToken`` call in the daemon thread of + ``_run_with_limit``, and return weak references to the copy and the + original. + + The thread function stores the copy in its closure. The traceback of the + original holds the frame of that function. + """ + original = ClientError(response("req-1"), "CreateMicrovmAuthToken") + options = SendJobOptions( + client=_TokenClient([original]), # type: ignore[arg-type] + microvm_id="mvm-1", + endpoint=endpoint_url, + path="/job", + port=8080, + body=b"{}", + retry_window_seconds=60, + remaining_seconds=lambda: 600.0, + logger=logging.getLogger("test"), + sleep=lambda seconds: None, + ) + copy: ClientError | None = None + try: + send_job(options) + except ClientError as error: + copy = error + assert copy is not None + assert copy.__cause__ is original + assert str(copy).endswith(" CreateMicrovmAuthToken request ID: req-1.") + # The remaining time sets a deadline. So the call ran in the daemon thread. + assert "_run_with_limit" in _frame_names(copy) + return weakref.ref(copy), weakref.ref(original) + + +def test_keeps_no_error_of_a_token_call_that_failed_under_a_deadline( + fake_endpoint: Any, +) -> None: + copy, original = _fail_a_token_call_under_a_deadline(fake_endpoint().url) + + gc.collect() + + assert copy() is None + assert original() is None + + +def _keep_the_copy_in_the_frame_of_the_caller() -> tuple[ + weakref.ref[ClientError], weakref.ref[ClientError] +]: + """Keep the copy in a local variable of the frame that called the failing + call, as a handler can in a closure. The traceback of the original holds + that frame.""" + original = ClientError(response(), "RunMicrovm") + + def call() -> None: + try: + raise original + except ClientError as error: + # The copy already has the error as its __cause__. + raise add_request_id("RunMicrovm", error) + + held: dict[str, ClientError] = {} + try: + call() + except ClientError as error: + held["copy"] = error + assert held["copy"].__cause__ is original + return weakref.ref(held["copy"]), weakref.ref(original) + + +def test_keeps_no_error_whose_traceback_reaches_the_copy() -> None: + copy, original = _keep_the_copy_in_the_frame_of_the_caller() + + gc.collect() + + assert copy() is None + assert original() is None + + +def test_gives_the_strategy_the_original_after_code_replaces_the_cause() -> None: + """Code can replace ``__cause__``. The copy holds the original in + ``_original_error`` too. So the strategy still receives the original.""" + error = ClientError(response(), "RunMicrovm") + labelled = add_request_id("RunMicrovm", error) + labelled.__cause__ = None + gc.collect() + seen: list[Exception] = [] + + def strategy(received: Exception, attempt: int) -> RetryDecision: + seen.append(received) + return RetryDecision.no_retry() + + decide_on_original_error(strategy)(labelled, 1) + + assert seen == [error] + assert seen[0] is error + + +def test_keeps_the_original_after_a_raise_from_another_error() -> None: + """``raise ... from ...`` sets ``__cause__``. It does not change + ``_original_error``.""" + error = ClientError(response(), "RunMicrovm") + labelled = add_request_id("RunMicrovm", error) + other = RuntimeError("other") + + with pytest.raises(ClientError) as raised: + raise labelled from other + + assert raised.value.__cause__ is other + seen: list[Exception] = [] + + def strategy(received: Exception, attempt: int) -> RetryDecision: + seen.append(received) + return RetryDecision.no_retry() + + decide_on_original_error(strategy)(raised.value, 1) + + assert seen[0] is error + + +def test_names_the_error_it_received_as_the_original_of_a_copy_of_a_copy() -> None: + """A copy of a copy takes ``__dict__`` of the first copy. The function + replaces ``_original_error`` with the first copy.""" + error = ClientError(response("req-1"), "RunMicrovm") + first = add_request_id("RunMicrovm", error) + + second = add_request_id("RunMicrovm", first) + + assert second is not first + assert vars(first)["_original_error"] is error + assert vars(second)["_original_error"] is first + assert second.__cause__ is first diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/request_ids_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/request_ids_test.py new file mode 100644 index 000000000..92a4bf8a0 --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/request_ids_test.py @@ -0,0 +1,1142 @@ +# SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. +# +# SPDX-License-Identifier: Apache-2.0 +"""Tests for the request IDs of failed AWS calls in MicroVM errors, run on +the local test runner.""" + +from __future__ import annotations + +import datetime +import functools +import gc +import logging +import re +import threading +import weakref +from collections import Counter +from collections.abc import Callable, Iterator +from dataclasses import dataclass +from typing import Any + +import boto3 +import botocore.session +import pytest +from aws_durable_execution_sdk_python.config import Duration +from aws_durable_execution_sdk_python.context import DurableContext +from aws_durable_execution_sdk_python.execution import ( + InvocationStatus, + durable_execution, +) +from aws_durable_execution_sdk_python.retries import ( + RetryDecision, + RetryStrategyConfig, + create_retry_strategy, +) +from aws_durable_execution_sdk_python_testing.runner import ( + DurableFunctionTestResult, + DurableFunctionTestRunner, +) +from botocore.credentials import CredentialProvider, RefreshableCredentials +from botocore.exceptions import ClientError +from botocore.stub import Stubber + +from aws_durable_execution_sdk_python_extras.microvm import ( + MicrovmConfig, + MicrovmDeliveryError, + MicrovmError, + MicrovmInvokeOptions, + MicrovmLaunchError, + MicrovmNotRunningError, + MicrovmRequestConfig, + MicrovmSessionConfig, + aws_errors, + microvm, + microvm_session, +) +from aws_durable_execution_sdk_python_extras.microvm import ( + operation as operation_module, +) + + +WAIT = 30 +REMOVED = ( + "MicroVM mvm-1 no longer exists. A MicroVM lives at most its `timeout` plus " + "5 minutes, and never longer than 8 hours." +) +DELIVERY = 'MicroVM "job": the job could not be delivered: ' +LAUNCH_FAILED = 'MicroVM "job": the launch failed: ' +STATE_ERROR = ( + "aws_durable_execution_sdk_python_extras.microvm.lifecycle.MicrovmStateError" +) +STS_DENIED = ( + "An error occurred (AccessDenied) when calling the AssumeRole operation: " + "not authorized to assume role" +) + + +@functools.cache +def _exceptions() -> Any: + """The modeled error classes of the lambda-microvms client.""" + return boto3.client("lambda-microvms", region_name="us-west-2").exceptions + + +def service_error( + code: str, + message: str, + operation: str, + request_id: str | None = "req-1", + status: int = 400, +) -> ClientError: + """An error of the class that botocore raises for ``code``, with the + ``response`` of a real answer.""" + metadata: dict[str, Any] = {"HTTPStatusCode": status} + if request_id is not None: + metadata["RequestId"] = request_id + response: Any = { + "Error": {"Code": code, "Message": message}, + "ResponseMetadata": metadata, + } + error: ClientError = _exceptions().from_code(code)(response, operation) + return error + + +def metadata(request_id: str | None) -> dict[str, Any]: + """The ``ResponseMetadata`` of an output.""" + if request_id is None: + return {"ResponseMetadata": {"HTTPStatusCode": 200}} + return {"ResponseMetadata": {"HTTPStatusCode": 200, "RequestId": request_id}} + + +Answer = BaseException | Callable[[], Any] | dict[str, Any] + + +class ScriptedClient: + """Answers each Lambda MicroVMs call from a list, and counts the calls. + + Each answer is an exception to raise, a function to call, or the output + to return as it is. After a list ends, each call returns a success. + """ + + def __init__(self, endpoint: str, **answers: list[Answer]) -> None: + self.endpoint = endpoint + self.calls: Counter[str] = Counter() + self._answers = {name: list(values) for name, values in answers.items()} + self._lock = threading.Lock() + + def _answer(self, name: str, success: dict[str, Any]) -> Any: + with self._lock: + self.calls[name] += 1 + answers = self._answers.get(name, []) + answer = answers.pop(0) if answers else success + if isinstance(answer, BaseException): + raise answer + if callable(answer): + return answer() + return answer + + def run_microvm(self, **request: Any) -> Any: + microvm = {"microvmId": "mvm-1", "state": "PENDING", "endpoint": self.endpoint} + return self._answer("run_microvm", microvm) + + def get_microvm(self, **request: Any) -> Any: + microvm = {"microvmId": "mvm-1", "state": "RUNNING", "endpoint": self.endpoint} + return self._answer("get_microvm", microvm) + + def resume_microvm(self, **request: Any) -> Any: + return self._answer("resume_microvm", {}) + + def create_microvm_auth_token(self, **request: Any) -> Any: + return self._answer("create_microvm_auth_token", {"authToken": {"x-t": "t"}}) + + def terminate_microvm(self, **request: Any) -> Any: + return self._answer("terminate_microvm", {}) + + +def base_fields(client: Any, strategy: Any) -> dict[str, Any]: + fields: dict[str, Any] = { + "image_identifier": "arn:aws:lambda:us-west-2:123456789012:microvm-image:img", + "execution_role_arn": "arn:aws:iam::123456789012:role/microvm", + "client": client, + } + if strategy is not None: + fields["retry_strategy"] = strategy + return fields + + +def catching( + kind: str, client: Any, caught: list[MicrovmError], strategy: Any = None +) -> Callable[..., Any]: + """A durable handler that runs one failing job and catches its error. + + ``kind`` is the operation: + + - ``microvm``: one job over HTTP, because the config sets a path. + - ``session``: a session with the default config. It makes no + ``GetMicrovm`` call. + - ``checking session``: a session with ``auto_suspend_on_idle``. It calls + ``GetMicrovm`` before each job. + + After the error, a durable wait ends the invocation. The next invocation + replays the failed operation from its checkpoint and catches the error + again. So ``caught`` holds the error of the first run and of the replay. + """ + fields = base_fields(client, strategy) + + @durable_execution + def handler(event: Any, context: DurableContext) -> Any: + try: + if kind == "microvm": + microvm( + context, + "job", + job_input=1, + config=MicrovmConfig( + **fields, + timeout=Duration.from_minutes(10), + request=MicrovmRequestConfig(path="/job"), + ), + ) + else: + microvm_session( + context, + "pipeline", + config=MicrovmSessionConfig( + **fields, + timeout=Duration.from_hours(1), + auto_suspend_on_idle=kind == "checking session", + ), + handler=lambda vm, ctx: vm.invoke( + ctx, + "job", + job_input=1, + options=MicrovmInvokeOptions( + path="/job", timeout=Duration.from_minutes(10) + ), + ), + ) + except MicrovmError as error: + caught.append(error) + else: + return "unexpected" + context.wait(Duration.from_seconds(1), name="replay") + return "caught" + + return handler + + +@pytest.fixture +def run() -> Iterator[Callable[..., DurableFunctionTestResult]]: + """Run a handler on the local test runner, and complete the named + callbacks with ``"ok"``.""" + runners: list[DurableFunctionTestRunner] = [] + + def _run( + handler: Callable[..., Any], callbacks: tuple[str, ...] = () + ) -> DurableFunctionTestResult: + runner = DurableFunctionTestRunner(handler=handler, poll_interval=0.05) + runners.append(runner) + arn = runner.run_async(input="null", execution_timeout=WAIT) + for name in callbacks: + callback_id = runner.wait_for_callback(arn, name=name, timeout=WAIT) + runner.send_callback_success(callback_id, b'"ok"') + return runner.wait_for_result(arn, timeout=WAIT) + + yield _run + for runner in runners: + runner.close() + + +def job_context(result: DurableFunctionTestResult, kind: str) -> Any: + """The child context of the job named ``job``.""" + if kind == "microvm": + return result.get_context("job") + return result.get_context("pipeline").get_context("job") + + +def assert_caught( + result: DurableFunctionTestResult, + caught: list[MicrovmError], + error_class: type[MicrovmError], + message: str, +) -> None: + """The handler caught ``error_class`` with ``message`` on the first run + and on replay.""" + assert result.status is InvocationStatus.SUCCEEDED, result.error + assert result.result == '"caught"' + assert len(caught) >= 2 + for error in caught: + assert type(error) is error_class + assert error.message == message + + +@dataclass(frozen=True) +class Scenario: + kind: str + answers: Callable[[str | None], dict[str, list[Answer]]] + """The answers of the client, for a request ID or ``None``.""" + error_class: type[MicrovmError] + prefix: str + step: str + step_error_type: str + step_message: tuple[str, str] + """The step message without and with the request ID ``req-1``.""" + call: str + calls: int + statuses: tuple[int, ...] = () + + +def _denied(operation: str) -> tuple[str, str]: + text = ( + f"An error occurred (AccessDeniedException) when calling the {operation} " + "operation: denied" + ) + return text, f"{text}. {operation} request ID: req-1." + + +SCENARIOS: dict[str, Scenario] = { + "CreateMicrovmAuthToken reports the MicroVM removed, for the first token " + "of a session job": Scenario( + kind="session", + answers=lambda rid: { + "create_microvm_auth_token": [ + service_error( + "ResourceNotFoundException", "gone", "CreateMicrovmAuthToken", rid + ) + ] + }, + error_class=MicrovmNotRunningError, + prefix=DELIVERY, + step="job.request", + step_error_type=STATE_ERROR, + step_message=(REMOVED, f"{REMOVED} CreateMicrovmAuthToken request ID: req-1."), + call="create_microvm_auth_token", + calls=1, + ), + "CreateMicrovmAuthToken reports the MicroVM removed, for the new token " + "after a 401": Scenario( + kind="session", + answers=lambda rid: { + "create_microvm_auth_token": [ + {"authToken": {"x-t": "t"}}, + service_error( + "ResourceNotFoundException", "gone", "CreateMicrovmAuthToken", rid + ), + ] + }, + error_class=MicrovmNotRunningError, + prefix=DELIVERY, + step="job.request", + step_error_type=STATE_ERROR, + step_message=(REMOVED, f"{REMOVED} CreateMicrovmAuthToken request ID: req-1."), + call="create_microvm_auth_token", + calls=2, + statuses=(401,), + ), + "CreateMicrovmAuthToken reports the MicroVM removed, in a session that " + "checks the state": Scenario( + kind="checking session", + answers=lambda rid: { + "create_microvm_auth_token": [ + service_error( + "ResourceNotFoundException", "gone", "CreateMicrovmAuthToken", rid + ) + ] + }, + error_class=MicrovmNotRunningError, + prefix=DELIVERY, + step="job.request", + step_error_type=STATE_ERROR, + step_message=(REMOVED, f"{REMOVED} CreateMicrovmAuthToken request ID: req-1."), + call="create_microvm_auth_token", + calls=1, + ), + "CreateMicrovmAuthToken reports the MicroVM removed, for a microvm job": Scenario( + kind="microvm", + answers=lambda rid: { + "create_microvm_auth_token": [ + service_error( + "ResourceNotFoundException", "gone", "CreateMicrovmAuthToken", rid + ) + ] + }, + error_class=MicrovmNotRunningError, + prefix=DELIVERY, + step="job.request", + step_error_type=STATE_ERROR, + step_message=(REMOVED, f"{REMOVED} CreateMicrovmAuthToken request ID: req-1."), + call="create_microvm_auth_token", + calls=1, + ), + "GetMicrovm reports the MicroVM removed": Scenario( + kind="checking session", + answers=lambda rid: { + "get_microvm": [ + service_error( + "ResourceNotFoundException", "gone", "GetMicrovm", rid, 404 + ) + ] + }, + error_class=MicrovmNotRunningError, + prefix=DELIVERY, + step="job.request", + step_error_type=STATE_ERROR, + step_message=(REMOVED, f"{REMOVED} GetMicrovm request ID: req-1."), + call="get_microvm", + calls=1, + ), + "ResumeMicrovm reports the MicroVM removed": Scenario( + kind="checking session", + answers=lambda rid: { + "get_microvm": [ + {"microvmId": "mvm-1", "state": "SUSPENDED", **metadata("get-1")} + ], + "resume_microvm": [ + service_error( + "ResourceNotFoundException", "gone", "ResumeMicrovm", rid, 404 + ) + ], + }, + error_class=MicrovmNotRunningError, + prefix=DELIVERY, + step="job.request", + step_error_type=STATE_ERROR, + step_message=(REMOVED, f"{REMOVED} ResumeMicrovm request ID: req-1."), + call="resume_microvm", + calls=1, + ), + # The request ID of an STS error belongs to STS. So the message names no + # RunMicrovm request ID, with or without a request ID on the error. + "the credential provider's STS call fails inside RunMicrovm": Scenario( + kind="microvm", + answers=lambda rid: { + "run_microvm": [ + ClientError( + { + "Error": { + "Code": "AccessDenied", + "Message": "not authorized to assume role", + }, + **metadata(rid), # type: ignore[typeddict-item] + }, + "AssumeRole", + ) + ] + }, + error_class=MicrovmLaunchError, + prefix=LAUNCH_FAILED, + step="job.launch", + step_error_type="botocore.exceptions.ClientError", + step_message=(STS_DENIED, STS_DENIED), + call="run_microvm", + calls=1, + ), + "GetMicrovm reports a terminated MicroVM": Scenario( + kind="checking session", + answers=lambda rid: { + "get_microvm": [ + { + "microvmId": "mvm-1", + "state": "TERMINATED", + "stateReason": "MaximumDuration", + **metadata(rid), + } + ] + }, + error_class=MicrovmNotRunningError, + prefix=DELIVERY, + step="job.request", + step_error_type=STATE_ERROR, + step_message=( + "MicroVM mvm-1 is TERMINATED: MaximumDuration.", + "MicroVM mvm-1 is TERMINATED: MaximumDuration. GetMicrovm request ID: " + "req-1.", + ), + call="get_microvm", + calls=1, + ), + "GetMicrovm reports a terminating MicroVM without a reason": Scenario( + kind="checking session", + answers=lambda rid: { + "get_microvm": [ + {"microvmId": "mvm-1", "state": "TERMINATING", **metadata(rid)} + ] + }, + error_class=MicrovmNotRunningError, + prefix=DELIVERY, + step="job.request", + step_error_type=STATE_ERROR, + step_message=( + "MicroVM mvm-1 is TERMINATING.", + "MicroVM mvm-1 is TERMINATING. GetMicrovm request ID: req-1.", + ), + call="get_microvm", + calls=1, + ), + "RunMicrovm rejects the launch": Scenario( + kind="microvm", + answers=lambda rid: { + "run_microvm": [ + service_error("ValidationException", "bad image", "RunMicrovm", rid) + ] + }, + error_class=MicrovmLaunchError, + prefix=LAUNCH_FAILED, + step="job.launch", + step_error_type="botocore.errorfactory.ValidationException", + step_message=( + "An error occurred (ValidationException) when calling the RunMicrovm " + "operation: bad image", + "An error occurred (ValidationException) when calling the RunMicrovm " + "operation: bad image. RunMicrovm request ID: req-1.", + ), + call="run_microvm", + calls=1, + ), + "RunMicrovm returns no microvmId": Scenario( + kind="microvm", + answers=lambda rid: {"run_microvm": [{"state": "PENDING", **metadata(rid)}]}, + error_class=MicrovmLaunchError, + prefix=LAUNCH_FAILED, + step="job.launch", + step_error_type="RuntimeError", + step_message=( + "RunMicrovm returned no microvmId", + "RunMicrovm returned no microvmId. RunMicrovm request ID: req-1.", + ), + call="run_microvm", + calls=1, + ), + "RunMicrovm returns no endpoint for HTTP delivery": Scenario( + kind="microvm", + answers=lambda rid: { + "run_microvm": [{"microvmId": "mvm-1", "state": "PENDING", **metadata(rid)}] + }, + error_class=MicrovmLaunchError, + prefix=LAUNCH_FAILED, + step="job.launch", + step_error_type="RuntimeError", + step_message=( + "RunMicrovm returned no endpoint", + "RunMicrovm returned no endpoint. RunMicrovm request ID: req-1.", + ), + call="run_microvm", + calls=1, + ), + "GetMicrovm denies access": Scenario( + kind="checking session", + answers=lambda rid: { + "get_microvm": [ + service_error("AccessDeniedException", "denied", "GetMicrovm", rid, 403) + ] + }, + error_class=MicrovmDeliveryError, + prefix=DELIVERY, + step="job.request", + step_error_type="botocore.errorfactory.AccessDeniedException", + step_message=_denied("GetMicrovm"), + call="get_microvm", + calls=1, + ), + "ResumeMicrovm denies access": Scenario( + kind="checking session", + answers=lambda rid: { + "get_microvm": [{"microvmId": "mvm-1", "state": "SUSPENDED"}], + "resume_microvm": [ + service_error( + "AccessDeniedException", "denied", "ResumeMicrovm", rid, 403 + ) + ], + }, + error_class=MicrovmDeliveryError, + prefix=DELIVERY, + step="job.request", + step_error_type="botocore.errorfactory.AccessDeniedException", + step_message=_denied("ResumeMicrovm"), + call="resume_microvm", + calls=1, + ), + "CreateMicrovmAuthToken denies access": Scenario( + kind="microvm", + answers=lambda rid: { + "create_microvm_auth_token": [ + service_error( + "AccessDeniedException", + "denied", + "CreateMicrovmAuthToken", + rid, + 403, + ) + ] + }, + error_class=MicrovmDeliveryError, + prefix=DELIVERY, + step="job.request", + step_error_type="botocore.errorfactory.AccessDeniedException", + step_message=_denied("CreateMicrovmAuthToken"), + call="create_microvm_auth_token", + calls=1, + ), + "CreateMicrovmAuthToken returns no authToken": Scenario( + kind="session", + answers=lambda rid: {"create_microvm_auth_token": [metadata(rid)]}, + error_class=MicrovmDeliveryError, + prefix=DELIVERY, + step="job.request", + step_error_type="RuntimeError", + step_message=( + "CreateMicrovmAuthToken returned no authToken", + "CreateMicrovmAuthToken returned no authToken. CreateMicrovmAuthToken " + "request ID: req-1.", + ), + call="create_microvm_auth_token", + calls=1, + ), +} + + +@pytest.mark.parametrize("label", list(SCENARIOS)) +@pytest.mark.parametrize( + ("request_id", "with_id"), [("req-1", 1), (None, 0)], ids=["with-id", "without-id"] +) +def test_shows_the_step_message_on_the_first_run_and_on_replay_without_a_retry( + run: Any, fake_endpoint: Any, label: str, request_id: str | None, with_id: int +) -> None: + scenario = SCENARIOS[label] + endpoint = fake_endpoint(list(scenario.statuses)) + client = ScriptedClient(endpoint.url, **scenario.answers(request_id)) + caught: list[MicrovmError] = [] + + result = run(catching(scenario.kind, client, caught)) + + step_message = scenario.step_message[with_id] + expected = f"{scenario.prefix}{step_message}" + assert_caught(result, caught, scenario.error_class, expected) + # The execution history keeps the same message. + context = job_context(result, scenario.kind) + step_error = context.get_step(scenario.step).error + assert step_error is not None + assert step_error.type == scenario.step_error_type + assert step_error.message == step_message + assert context.error is not None + assert context.error.message == expected + # The default retry strategy fails the step at once. + assert client.calls[scenario.call] == scenario.calls + + +def test_names_the_request_id_that_botocore_parses_from_a_real_answer( + run: Any, fake_endpoint: Any +) -> None: + """A real lambda-microvms client, with botocore's Stubber, raises the + modeled error with the request ID in ``ResponseMetadata``.""" + endpoint = fake_endpoint() + client = boto3.session.Session().client("lambda-microvms", region_name="us-west-2") + started = datetime.datetime(2026, 1, 1, tzinfo=datetime.UTC) + caught: list[MicrovmError] = [] + with Stubber(client) as stub: + stub.add_response( + "run_microvm", + { + "microvmId": "mvm-1", + "state": "PENDING", + "endpoint": endpoint.url, + "imageArn": "arn:aws:lambda:us-west-2:123456789012:microvm-image:img", + "imageVersion": "1", + "maximumDurationInSeconds": 3900, + "startedAt": started, + }, + ) + stub.add_client_error( + "get_microvm", + service_error_code="ResourceNotFoundException", + service_message="MicroVM not found: mvm-1", + http_status_code=404, + response_meta={"RequestId": "real-1"}, + ) + stub.add_response("terminate_microvm", {}) + + result = run(catching("checking session", client, caught)) + + stub.assert_no_pending_responses() + client.close() + assert_caught( + result, + caught, + MicrovmNotRunningError, + f"{DELIVERY}{REMOVED} GetMicrovm request ID: real-1.", + ) + + +@pytest.mark.parametrize("request_id", ["sts-1", None], ids=["with-id", "without-id"]) +def test_names_no_request_id_for_an_sts_error_of_a_real_credential_provider( + run: Any, request_id: str | None +) -> None: + """botocore resolves the credentials inside the RunMicrovm call. A + role-assuming provider calls STS there, and STS denies the role.""" + sts = boto3.client( + "sts", + region_name="us-west-2", + aws_access_key_id="key", + aws_secret_access_key="secret", + ) + stub = Stubber(sts) + stub.add_client_error( + "assume_role", + service_error_code="AccessDenied", + service_message="not authorized to assume role", + http_status_code=403, + response_meta={} if request_id is None else {"RequestId": request_id}, + ) + stub.activate() + + def assume_role() -> dict[str, str]: + sts.assume_role( + RoleArn="arn:aws:iam::123456789012:role/r", RoleSessionName="job" + ) + raise AssertionError # pragma: no cover - STS denies the role + + expired = datetime.datetime.now(datetime.UTC) - datetime.timedelta(hours=1) + + class AssumeRoleProvider(CredentialProvider): + """Returns expired credentials. botocore refreshes them with + ``assume_role`` when it signs the next request.""" + + METHOD = "assume-role" + + def load(self) -> RefreshableCredentials: + return RefreshableCredentials.create_from_metadata( + metadata={ + "access_key": "key", + "secret_key": "secret", + "token": "token", + "expiry_time": expired.isoformat(), + }, + refresh_using=assume_role, + method=self.METHOD, + ) + + session = botocore.session.get_session() + session.get_component("credential_provider").insert_before( + "env", AssumeRoleProvider() + ) + client = session.create_client("lambda-microvms", region_name="us-west-2") + launches: list[str] = [] + client.meta.events.register( + "before-parameter-build.lambda-microvms.RunMicrovm", + lambda **kwargs: launches.append("RunMicrovm"), + ) + caught: list[MicrovmError] = [] + + result = run(catching("microvm", client, caught)) + + stub.assert_no_pending_responses() + client.close() + assert_caught(result, caught, MicrovmLaunchError, f"{LAUNCH_FAILED}{STS_DENIED}") + step_error = result.get_context("job").get_step("job.launch").error + assert step_error is not None + assert step_error.type == "botocore.exceptions.ClientError" + assert step_error.message == STS_DENIED + # The default retry strategy does not retry AccessDenied. + assert launches == ["RunMicrovm"] + + +def throttled(operation: str, *, with_id: bool, count: int = 5) -> list[Answer]: + """A new ThrottlingException for each call, as botocore raises it.""" + return [ + service_error( + "ThrottlingException", + "Rate exceeded", + operation, + f"req-{i + 1}" if with_id else None, + 429, + ) + for i in range(count) + ] + + +def _throttle_text(operation: str) -> str: + return ( + f"An error occurred (ThrottlingException) when calling the {operation} " + "operation: Rate exceeded" + ) + + +@pytest.mark.parametrize("with_id", [True, False], ids=["with-id", "without-id"]) +def test_retries_a_throttled_auth_token_5_times_and_names_the_last_call( + run: Any, fake_endpoint: Any, with_id: bool +) -> None: + endpoint = fake_endpoint() + tokens = throttled("CreateMicrovmAuthToken", with_id=with_id) + client = ScriptedClient(endpoint.url, create_microvm_auth_token=tokens) + caught: list[MicrovmError] = [] + + result = run(catching("session", client, caught)) + + text = _throttle_text("CreateMicrovmAuthToken") + message = f"{text}. CreateMicrovmAuthToken request ID: req-5." if with_id else text + assert_caught(result, caught, MicrovmDeliveryError, f"{DELIVERY}{message}") + assert client.calls["create_microvm_auth_token"] == 5 + assert endpoint.requests == [] + + +@pytest.mark.parametrize("with_id", [True, False], ids=["with-id", "without-id"]) +def test_retries_a_throttled_launch_5_times_and_names_the_last_call( + run: Any, fake_endpoint: Any, with_id: bool +) -> None: + launches = throttled("RunMicrovm", with_id=with_id) + client = ScriptedClient(fake_endpoint().url, run_microvm=launches) + caught: list[MicrovmError] = [] + + result = run(catching("microvm", client, caught)) + + text = _throttle_text("RunMicrovm") + message = f"{text}. RunMicrovm request ID: req-5." if with_id else text + assert_caught(result, caught, MicrovmLaunchError, f"{LAUNCH_FAILED}{message}") + assert client.calls["run_microvm"] == 5 + + +def test_names_the_request_id_once_for_one_error_object_raised_by_every_call( + run: Any, fake_endpoint: Any +) -> None: + error = service_error("ThrottlingException", "Rate exceeded", "RunMicrovm") + client = ScriptedClient(fake_endpoint().url, run_microvm=[error] * 5) + caught: list[MicrovmError] = [] + + result = run(catching("microvm", client, caught)) + + text = _throttle_text("RunMicrovm") + assert_caught( + result, + caught, + MicrovmLaunchError, + f"{LAUNCH_FAILED}{text}. RunMicrovm request ID: req-1.", + ) + assert client.calls["run_microvm"] == 5 + # The error that the client raised keeps its message. + assert str(error) == text + + +def test_names_only_the_last_request_id_of_one_error_object( + run: Any, fake_endpoint: Any +) -> None: + error = service_error("ThrottlingException", "Rate exceeded", "RunMicrovm", None) + calls = iter(range(1, 6)) + + def raise_with_new_id() -> Any: + error.response["ResponseMetadata"]["RequestId"] = f"r-{next(calls)}" + raise error + + client = ScriptedClient(fake_endpoint().url, run_microvm=[raise_with_new_id] * 5) + caught: list[MicrovmError] = [] + + result = run(catching("microvm", client, caught)) + + message = f"{_throttle_text('RunMicrovm')}. RunMicrovm request ID: r-5." + assert_caught(result, caught, MicrovmLaunchError, f"{LAUNCH_FAILED}{message}") + step_error = result.get_context("job").get_step("job.launch").error + assert step_error is not None + assert step_error.message == message + assert client.calls["run_microvm"] == 5 + + +def test_gives_a_custom_strategy_the_error_that_botocore_raised( + run: Any, fake_endpoint: Any +) -> None: + tokens = throttled("CreateMicrovmAuthToken", with_id=True) + client = ScriptedClient(fake_endpoint().url, create_microvm_auth_token=tokens) + seen: list[tuple[Exception, str]] = [] + + def strategy(error: Exception, attempt: int) -> RetryDecision: + seen.append((error, str(error))) + if attempt < 2: + return RetryDecision.retry(Duration.from_seconds(1)) + return RetryDecision.no_retry() + + caught: list[MicrovmError] = [] + + result = run(catching("session", client, caught, strategy)) + + # The launch succeeded, so the strategy saw only the two token errors. + text = _throttle_text("CreateMicrovmAuthToken") + assert [error for error, _ in seen] == tokens[:2] + for (error, message), raised in zip(seen, tokens, strict=False): + assert error is raised + assert message == text + assert_caught( + result, + caught, + MicrovmDeliveryError, + f"{DELIVERY}{text}. CreateMicrovmAuthToken request ID: req-2.", + ) + + +def _fail_five_tokens_of_a_session_job( + endpoint_url: str, monkeypatch: pytest.MonkeyPatch +) -> tuple[list[weakref.ref[ClientError]], list[weakref.ref[ClientError]]]: + """Run a session job whose 5 ``CreateMicrovmAuthToken`` calls are + throttled. The handler catches the error and keeps no reference to it. + + Return weak references to the copies and to the originals. + + 1. The core SDK calls the step retry strategy with the copy. + 2. :func:`decide_on_original_error` passes the original on to the + strategy in the config. + 3. So the test wraps :func:`decide_on_original_error` where the + operation calls it, and records each copy that it receives. + """ + tokens = throttled("CreateMicrovmAuthToken", with_id=True) + originals: list[weakref.ref[ClientError]] = [ + weakref.ref(error) for error in tokens if isinstance(error, ClientError) + ] + client = ScriptedClient(endpoint_url, create_microvm_auth_token=tokens) + del tokens + copies: list[weakref.ref[ClientError]] = [] + threaded: list[bool] = [] + + def recording(strategy: Any) -> Any: + decide = aws_errors.decide_on_original_error(strategy) + + def record(error: Exception, attempt: int) -> RetryDecision: + if isinstance(error, ClientError) and "_original_error" in vars(error): + copies.append(weakref.ref(error)) + threaded.append("_run_with_limit" in _frame_names(error.__traceback__)) + return decide(error, attempt) + + return record + + monkeypatch.setattr(operation_module, "decide_on_original_error", recording) + fields = base_fields(client, None) + + @durable_execution + def handler(event: Any, context: DurableContext) -> Any: + try: + microvm_session( + context, + "pipeline", + config=MicrovmSessionConfig(**fields, timeout=Duration.from_hours(1)), + handler=lambda vm, ctx: vm.invoke( + ctx, + "job", + job_input=1, + options=MicrovmInvokeOptions( + path="/job", timeout=Duration.from_minutes(10) + ), + ), + ) + except MicrovmError: + return "caught" + return "unexpected" # pragma: no cover - the job fails + + runner = DurableFunctionTestRunner(handler=handler, poll_interval=0.05) + try: + arn = runner.run_async(input="null", execution_timeout=WAIT) + result = runner.wait_for_result(arn, timeout=WAIT) + finally: + runner.close() + assert result.status is InvocationStatus.SUCCEEDED, result.error + assert result.result == '"caught"' + assert client.calls["create_microvm_auth_token"] == 5 + assert len(copies) == 5 + # The local test runner gives the invocation a remaining time. So each + # call ran in the daemon thread of _run_with_limit. + assert threaded == [True] * 5 + return copies, originals + + +def _frame_names(traceback: Any) -> list[str]: + names = [] + while traceback is not None: + names.append(traceback.tb_frame.f_code.co_name) + traceback = traceback.tb_next + return names + + +def test_keeps_no_error_of_a_throttled_token_call_after_the_execution( + fake_endpoint: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + """1. The traceback of each original holds the frame of the thread + function of ``_run_with_limit``. That frame holds the copy. + 2. The copy holds the original in ``_original_error``. So the copy and + the original refer to each other. + 3. No module state holds either error. So after the execution, the + garbage collector must free the copies and the originals. + + The core SDK logs each failed step attempt with the error. pytest keeps + each log record until the test ends, and a record holds its error. So + the test turns off the step logger. + """ + step_logger = logging.getLogger("aws_durable_execution_sdk_python.operation.step") + monkeypatch.setattr(step_logger, "disabled", True) + copies, originals = _fail_five_tokens_of_a_session_job( + fake_endpoint().url, monkeypatch + ) + + gc.collect() + + assert [copy() for copy in copies] == [None] * 5 + assert [original() for original in originals] == [None] * 5 + + +@pytest.mark.parametrize( + ("request_id", "message"), + [ + ( + "req-1", + "An error occurred (AccessDeniedException) when calling the " + "TerminateMicrovm operation: denied. TerminateMicrovm request ID: req-1.", + ), + ( + None, + "An error occurred (AccessDeniedException) when calling the " + "TerminateMicrovm operation: denied", + ), + ], + ids=["with-id", "without-id"], +) +def test_keeps_the_job_result_and_records_the_terminate_message( + run: Any, fake_endpoint: Any, request_id: str | None, message: str +) -> None: + denied = service_error( + "AccessDeniedException", "denied", "TerminateMicrovm", request_id, 403 + ) + client = ScriptedClient(fake_endpoint().url, terminate_microvm=[denied]) + + @durable_execution + def handler(event: Any, context: DurableContext) -> Any: + return microvm( + context, + "build", + job_input=1, + config=MicrovmConfig( + **base_fields(client, None), timeout=Duration.from_minutes(10) + ), + ) + + result = run(handler, ("build.callback",)) + + assert result.status is InvocationStatus.SUCCEEDED, result.error + assert result.result == '"ok"' + assert client.calls["terminate_microvm"] == 1 + step_error = result.get_context("build").get_step("build.terminate").error + assert step_error is not None + assert step_error.type == "botocore.errorfactory.AccessDeniedException" + assert step_error.message == message + + +def test_names_no_terminate_request_id_for_an_error_of_another_operation( + run: Any, fake_endpoint: Any +) -> None: + """A custom client raises one error object from CreateMicrovmAuthToken + and then from TerminateMicrovm. The error names CreateMicrovmAuthToken + as its operation. So the terminate record names no request ID: it would + be the ID of another call.""" + shared = service_error( + "AccessDeniedException", "denied", "CreateMicrovmAuthToken", "shared-1", 403 + ) + client = ScriptedClient( + fake_endpoint().url, + create_microvm_auth_token=[shared], + terminate_microvm=[shared], + ) + caught: list[MicrovmError] = [] + + result = run(catching("session", client, caught)) + + text = ( + "An error occurred (AccessDeniedException) when calling the " + "CreateMicrovmAuthToken operation: denied" + ) + assert_caught( + result, + caught, + MicrovmDeliveryError, + f"{DELIVERY}{text}. CreateMicrovmAuthToken request ID: shared-1.", + ) + terminate = result.get_context("pipeline").get_step("pipeline.terminate").error + assert terminate is not None + assert terminate.message == text + assert str(shared) == text + + +@pytest.mark.parametrize( + ("call", "kind", "pattern", "calls"), + [ + (call, kind, pattern, calls) + for call, kind in [ + ("run_microvm", "microvm"), + ("create_microvm_auth_token", "session"), + ("terminate_microvm", "terminate"), + ("get_microvm", "checking session"), + ] + for pattern, calls in [ + (re.compile(r": Rate exceeded$"), 3), + (re.compile(r"request", re.IGNORECASE), 1), + ] + ], + ids=lambda value: value.pattern if isinstance(value, re.Pattern) else str(value), +) +def test_a_message_based_strategy_decides_on_the_message_of_the_service( + run: Any, + fake_endpoint: Any, + call: str, + kind: str, + pattern: re.Pattern[str], + calls: int, +) -> None: + """``create_retry_strategy`` matches its patterns against ``str(error)``. + + 1. The anchored pattern matches the botocore message. So a throttled call + is retried until the third attempt. + 2. The loose pattern matches no botocore message of a throttled call. So + the call is not retried. + 3. The step message has the request ID. A strategy that saw it would + decide the other way in both cases. + """ + operation = { + "run_microvm": "RunMicrovm", + "create_microvm_auth_token": "CreateMicrovmAuthToken", + "terminate_microvm": "TerminateMicrovm", + "get_microvm": "GetMicrovm", + }[call] + client = ScriptedClient( + fake_endpoint().url, **{call: throttled(operation, with_id=True, count=3)} + ) + strategy = create_retry_strategy( + RetryStrategyConfig( + max_attempts=3, + initial_delay=Duration.from_seconds(1), + retryable_errors=[pattern], + ) + ) + caught: list[MicrovmError] = [] + if kind == "terminate": + + @durable_execution + def handler(event: Any, context: DurableContext) -> Any: + return microvm( + context, + "job", + job_input=1, + config=MicrovmConfig( + **base_fields(client, strategy), timeout=Duration.from_minutes(10) + ), + ) + + result = run(handler, ("job.callback",)) + else: + result = run(catching(kind, client, caught, strategy)) + + # The decision first. It is the same as before the request ID existed. + assert client.calls[call] == calls + message = f"{_throttle_text(operation)}. {operation} request ID: req-{calls}." + if kind == "terminate": + assert result.status is InvocationStatus.SUCCEEDED, result.error + assert result.result == '"ok"' + record = result.get_context("job").get_step("job.terminate").error + assert record is not None + assert record.message == message + else: + prefix = LAUNCH_FAILED if call == "run_microvm" else DELIVERY + error_class = ( + MicrovmLaunchError if call == "run_microvm" else MicrovmDeliveryError + ) + assert_caught(result, caught, error_class, f"{prefix}{message}")