diff --git a/packages/aws-durable-execution-sdk-python-extras/README.md b/packages/aws-durable-execution-sdk-python-extras/README.md index bfc9e463e..c30f30ce5 100644 --- a/packages/aws-durable-execution-sdk-python-extras/README.md +++ b/packages/aws-durable-execution-sdk-python-extras/README.md @@ -236,16 +236,18 @@ The handler's return value is checkpointed as the session's result. So it must b A running MicroVM pays compute charges. A suspended MicroVM pays only for snapshot storage, and it keeps its memory and files. A session can suspend its MicroVM between jobs in two ways: -1. `auto_suspend_on_idle=True` asks the worker in the MicroVM to suspend its own MicroVM when no job has run for `auto_suspend_idle_time`. The default idle time is 60 seconds, or the session `timeout` if that is shorter. The idle time must be at least 10 seconds. +1. `auto_suspend_on_idle=True` asks the worker in the MicroVM to suspend its own MicroVM when no job has run for `auto_suspend_idle_time`. The default idle time is 60 seconds, or the session `timeout` if that is shorter. The idle time must be at least 10 seconds. So a session whose `timeout` is under 10 seconds does not suspend by default, even with `auto_suspend_on_idle=True`. 2. `idle_policy` sets the `RunMicrovm` idle policy. The service then suspends the MicroVM when its endpoint receives no traffic. A running job receives no traffic after its delivery. So `maxIdleDurationSeconds` must be longer than the longest job. -With either setting, each request step calls `GetMicrovm` before it sends the job. It resumes a suspended MicroVM, and it sends the job to the endpoint that `GetMicrovm` reports. The MicroVM can suspend itself between that check and the request. The endpoint then answers 502, or the worker answers 503. So after a 502, 503, or 504, or a request without an answer, the step checks the state again before it retries. The wait for a resume counts toward the request's `retry_window`. +For a session that launched with either setting, each request step calls `GetMicrovm` before it sends the job. It resumes a suspended MicroVM, and it sends the job to the endpoint that `GetMicrovm` reports. The MicroVM can suspend itself between that check and the request. The endpoint then answers 502, or the worker answers 503. So after a 502, 503, or 504, or a request without an answer, the step checks the state again before it retries. The wait for a resume counts toward the request's `retry_window`. -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. Without either setting, the session 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. +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. Suspending is off by default, because it needs more permissions and changes what the MicroVM does between jobs: -- The function's role needs `lambda:GetMicrovm` and `lambda:ResumeMicrovm`. Without them, every job fails. +- The function's role needs `lambda:GetMicrovm` and `lambda:ResumeMicrovm`. Without them, every job fails. A later deploy that turns suspending off does not change the resume mode of a session that launched with it. So keep `lambda:GetMicrovm` and `lambda:ResumeMicrovm` until every session that launched with `auto_suspend_on_idle=True` or an `idle_policy` has ended. - With `auto_suspend_on_idle`, the MicroVM's role needs `lambda:SuspendMicrovm` on the image. That permission lets code in any MicroVM of the image suspend any other. So use one image per trust boundary. - Background processes that an earlier job started stop while the MicroVM is suspended. The worker counts only its running jobs. 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 116a3dbcc..2d4a926e9 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 @@ -134,8 +134,9 @@ class LaunchResult: """What the launch step checkpoints. - Wire names: ``microvmId``, ``endpoint``, and ``delivery``. The JavaScript - operation checkpoints the same document. + Wire names: ``microvmId``, ``endpoint``, ``delivery``, and + ``resumeBeforeDelivery``. The JavaScript operation checkpoints the same + document. """ microvm_id: str @@ -144,15 +145,29 @@ class LaunchResult: so every replay follows the delivery that the launch prepared for.""" endpoint: str | None = None """The MicroVM's endpoint. Present only for HTTP delivery.""" + resume_before_delivery: bool | None = None + """Whether each session job checks the MicroVM state, and resumes a + suspended MicroVM, before delivery. Only a session launch records it. + Replay reads it from the checkpoint. So every replay follows the mode that + the launch prepared the MicroVM for. A launch checkpoint from code that + predates this field does not have it.""" @classmethod def from_dict(cls, data: dict[str, Any]) -> LaunchResult: # A checkpoint without a delivery gets an empty one. The operation # then terminates the MicroVM as for any delivery it does not know. + # Only a bool is a resume mode. In Python True == 1, so a check by + # equality would accept 1. Any other value reads as no resume mode. + resume_before_delivery = data.get("resumeBeforeDelivery") return cls( microvm_id=data["microvmId"], delivery=data.get("delivery", ""), endpoint=data.get("endpoint"), + resume_before_delivery=( + resume_before_delivery + if isinstance(resume_before_delivery, bool) + else None + ), ) def to_dict(self) -> dict[str, Any]: @@ -160,6 +175,10 @@ def to_dict(self) -> dict[str, Any]: if self.endpoint is not None: data["endpoint"] = self.endpoint data["delivery"] = self.delivery + # A microvm() launch records no resume mode. So its checkpoint has no + # such key, as in the JavaScript operation. + if self.resume_before_delivery is not None: + data["resumeBeforeDelivery"] = self.resume_before_delivery return data @@ -363,12 +382,16 @@ def launch( payload: str, delivery: str, idle_policy: Mapping[str, Any] | None = None, + resume_before_delivery: bool | None = None, ) -> LaunchResult: """Run the launch step, ``.launch``. The client token is the SHA-256 hex digest of ``token_source``. It must be unique per launch and stable across retries and replays. + ``resume_before_delivery`` is the resume mode to record. Only a + session launch sets it. + Raises: MicrovmLaunchError: When ``RunMicrovm`` fails after all retries. """ @@ -376,7 +399,11 @@ def launch( return LaunchResult.from_dict( child.step( lambda _: self._launch_call( - token_source, payload, delivery, idle_policy + token_source, + payload, + delivery, + idle_policy, + resume_before_delivery, ).to_dict(), name=f"{self.name}.launch", config=StepConfig( @@ -540,6 +567,7 @@ def _launch_call( payload: str, delivery: str, idle_policy: Mapping[str, Any] | None, + resume_before_delivery: bool | None = None, ) -> LaunchResult: """Call ``RunMicrovm``, and return what the launch step checkpoints.""" over_http = delivery == HTTP_DELIVERY @@ -589,6 +617,7 @@ def _launch_call( microvm_id=microvm_id, delivery=delivery, endpoint=endpoint if over_http else None, + resume_before_delivery=resume_before_delivery, ) def _terminate_call(self, microvm_id: str) -> None: 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 591048b83..3bffbfab9 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 @@ -106,7 +106,9 @@ def microvm_session( its own MicroVM when no job has run for ``auto_suspend_idle_time``. With that setting or an ``idle_policy``, the request step of each job calls ``GetMicrovm`` first. It resumes a suspended MicroVM, and it returns at - once for a new MicroVM that is still booting. + once for a new MicroVM that is still booting. The launch checkpoint + records whether the MicroVM can be suspended. Replay follows that record, + not the current config. A failed :meth:`MicrovmSession.invoke` raises inside the handler. The handler can catch the error and continue with the same MicroVM. An @@ -198,9 +200,16 @@ class MicrovmSession: ``map`` and ``parallel`` items. """ - def __init__(self, scope: _SessionScope, launched: LaunchResult) -> None: + def __init__( + self, + scope: _SessionScope, + launched: LaunchResult, + *, + resume_before_delivery: bool, + ) -> None: self._scope = scope self._launched = launched + self._resume_before_delivery = resume_before_delivery @property def microvm_id(self) -> str: @@ -283,6 +292,7 @@ def invoke( heartbeat_timeout_seconds = _validate_heartbeat(name, options.heartbeat_timeout) _validate_input(name, job_input) launched = self._launched + resume = self._resume_before_delivery def run(job_context: DurableContext) -> Any: callback = create_job_callback( @@ -297,7 +307,7 @@ def run(job_context: DurableContext) -> Any: heartbeat_timeout_seconds=heartbeat_timeout_seconds, request=options, retry_window_seconds=retry_window_seconds, - resume=scope.resume_before_delivery, + resume=resume, ) try: return callback.result() @@ -327,11 +337,13 @@ class _SessionScope(_MicrovmScope): in it would be outside the session's child context.""" @property - def resume_before_delivery(self) -> bool: - """Whether each job must check the MicroVM state, and resume it. + def resume_from_config(self) -> bool: + """Whether the current config asks each job to check the MicroVM + state, and resume it. The worker suspends an idle MicroVM, and an idle policy can suspend - it too. Either way, each job must resume it first. + it too. Either way, each job must resume it first. A session launch + records this value as its resume mode. """ return ( self.auto_suspend_idle_seconds is not None or self.idle_policy is not None @@ -381,14 +393,38 @@ def run_session( ), delivery=HTTP_DELIVERY, idle_policy=self.idle_policy, + resume_before_delivery=self.resume_from_config, ) if launched.delivery != HTTP_DELIVERY: # A launch checkpoint from a release with another delivery. Every # session job goes over HTTP, so this release cannot run one. self.terminate(child, launched.microvm_id) raise unknown_delivery_error(self.name, launched) + # 1. The worker reads its idle time once, from the run hook payload of + # the launch. RunMicrovm sets the idle policy once, at the launch. + # 2. The launch step's result comes from the checkpoint on replay. + # 3. So a deploy that changes auto_suspend_on_idle or idle_policy does + # not change what the running MicroVM does. + # 4. A job that skips the state check of a suspended MicroVM gets a + # 502. The request step retries a 502 until both retry tiers end. + # 5. So each job follows the resume mode that the launch recorded, + # not the current config. + # A launch checkpoint from code that predates the resumeBeforeDelivery + # field has no such field. That code computed the mode from the + # current config on every replay. So the session does the same for + # such a checkpoint. + resume_before_delivery = ( + launched.resume_before_delivery + if launched.resume_before_delivery is not None + else self.resume_from_config + ) try: - result = handler(MicrovmSession(self, launched), child) + result = handler( + MicrovmSession( + self, launched, resume_before_delivery=resume_before_delivery + ), + child, + ) except ExecutionError: # A replay that does not match the history raises # NonDeterministicExecutionError, and the execution fails without diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py index 6b45d9866..12d977801 100644 --- a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py @@ -463,6 +463,48 @@ def test_sends_every_job_over_http_to_a_configured_route( assert client.requests("create_microvm_auth_token")[0]["allowedPorts"] == [ {"port": 3000} ] + # A single job never resumes its MicroVM. So its launch checkpoint has no + # resume mode. + launch = result.get_context("build").get_step("build.launch") + assert launch.result is not None + assert json.loads(launch.result) == { + "microvmId": "mvm-1", + "endpoint": endpoint.url, + "delivery": "http", + } + + +@pytest.mark.parametrize("resume_before_delivery", [True, False]) +def test_launch_result_writes_and_reads_a_resume_mode( + resume_before_delivery: bool, # noqa: FBT001 - a pytest parameter +) -> None: + launched = operation.LaunchResult( + microvm_id="mvm-1", + delivery="http", + endpoint="mvm-1.lambda-microvm.us-east-1.on.aws", + resume_before_delivery=resume_before_delivery, + ) + + data = launched.to_dict() + + assert data == { + "microvmId": "mvm-1", + "endpoint": "mvm-1.lambda-microvm.us-east-1.on.aws", + "delivery": "http", + "resumeBeforeDelivery": resume_before_delivery, + } + assert operation.LaunchResult.from_dict(json.loads(json.dumps(data))) == launched + + +@pytest.mark.parametrize("value", ["false", "true", 1, 0, None, [], {}]) +def test_launch_result_reads_only_a_bool_as_a_resume_mode(value: Any) -> None: + # In Python True == 1. So a check by equality would read 1 as True. + launched = operation.LaunchResult.from_dict( + {"microvmId": "mvm-1", "delivery": "http", "resumeBeforeDelivery": value} + ) + + assert launched.resume_before_delivery is None + assert "resumeBeforeDelivery" not in launched.to_dict() def test_keeps_the_configured_ingress_for_http( diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py index 2e914e265..0fa3f5c0e 100644 --- a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py @@ -9,6 +9,7 @@ import hashlib import json from collections.abc import Callable, Iterator +from types import SimpleNamespace from typing import Any import boto3 @@ -1148,3 +1149,211 @@ def body(vm: MicrovmSession, ctx: DurableContext, event: Any) -> Any: assert result.error is not None assert "history does not match" in (result.error.message or "") assert client.requests("terminate_microvm") == [] + + +IDLE_POLICY = { + "maxIdleDurationSeconds": 60, + "suspendedDurationSeconds": 3600, + "autoResumeEnabled": False, +} + + +def launch_checkpoint(result: DurableFunctionTestResult) -> dict[str, Any]: + """The document that the session's launch step checkpointed.""" + launch = result.get_context("pipeline").get_step("pipeline.launch") + assert launch.result is not None + checkpoint: dict[str, Any] = json.loads(launch.result) + return checkpoint + + +def handler_across_deploy( + client: Any, + first: dict[str, Any], + later: dict[str, Any], + body: Callable[[MicrovmSession, DurableContext, Any], Any], +) -> tuple[Callable[..., Any], list[int]]: + """A durable handler whose first invocation uses the session config + ``first``, and every later invocation uses ``later``. So it behaves as a + deploy between the launch and a replay. The list counts the invocations. + """ + invocations: list[int] = [] + + @durable_execution + def handler(event: Any, context: DurableContext) -> Any: + invocations.append(len(invocations) + 1) + fields = first if len(invocations) == 1 else later + try: + return { + "result": microvm_session( + context, + "pipeline", + config=session_config(client, **fields), + handler=lambda vm, ctx: body(vm, ctx, event), + ) + } + except MicrovmError as error: + return {"error": type(error).__name__, "message": error.message} + + return handler, invocations + + +def checkpoint_launch_as(monkeypatch: pytest.MonkeyPatch, **fields: Any) -> None: + """Make the launch step checkpoint the MicroVM fields of a real launch, + plus ``fields``, and no resume mode of its own. So a test controls the + launch checkpoint that every replay reads.""" + original = operation._MicrovmScope._launch_call # noqa: SLF001 + + def launch(self: Any, *args: Any) -> Any: + data = original(self, *args).to_dict() + data.pop("resumeBeforeDelivery", None) + data.update(fields) + return SimpleNamespace(to_dict=lambda: data) + + monkeypatch.setattr(operation._MicrovmScope, "_launch_call", launch) # noqa: SLF001 + + +@pytest.mark.parametrize( + ("fields", "resume_before_delivery"), + [ + ({"auto_suspend_on_idle": True}, True), + ({"idle_policy": IDLE_POLICY}, True), + ({}, False), + ], + ids=["auto_suspend_on_idle", "idle_policy", "neither"], +) +def test_records_the_resume_mode_in_the_launch_checkpoint( + run: Any, + fake_client: Any, + fields: dict[str, Any], + resume_before_delivery: bool, # noqa: FBT001 - a pytest parameter +) -> None: + client = fake_client() + + result = run( + session_handler(session_config(client, **fields), lambda vm, ctx, e: None) + ) + + assert outcome(result) == {"result": None} + assert launch_checkpoint(result) == { + "microvmId": "mvm-1", + "endpoint": client.endpoint, + "delivery": "http", + "resumeBeforeDelivery": resume_before_delivery, + } + + +@pytest.mark.parametrize( + "first", + [{"auto_suspend_on_idle": True}, {"idle_policy": IDLE_POLICY}], + ids=["auto_suspend_on_idle", "idle_policy"], +) +def test_still_resumes_before_each_job_when_a_deploy_turns_suspending_off( + run: Any, fake_client: Any, fake_endpoint: Any, first: dict[str, Any] +) -> None: + # The worker reads its idle time from the run hook payload of the launch. + # RunMicrovm sets the idle policy at the launch. So the MicroVM still + # suspends during the first job's wait, after the deploy. + endpoint = fake_endpoint() + client = fake_client( + endpoint=endpoint.url, states=["RUNNING", "SUSPENDED", "RUNNING"] + ) + handler, invocations = handler_across_deploy(client, first, {}, two_jobs) + + result = run(handler, None, [("build", succeed_with(1)), ("test", succeed_with(2))]) + + assert outcome(result)["result"]["test"] == 2 + assert len(invocations) > 1 + # The second job runs after the deploy. It checks the state, and resumes + # the MicroVM before delivery. + assert len(client.requests("get_microvm")) == 3 + assert client.requests("resume_microvm") == [{"microvmIdentifier": "mvm-1"}] + assert len(endpoint.requests) == 2 + assert launch_checkpoint(result)["resumeBeforeDelivery"] is True + + +def test_makes_no_state_check_when_a_deploy_turns_suspending_on( + run: Any, fake_client: Any, fake_endpoint: Any +) -> None: + # The launch asked the worker for no suspend and set no idle policy. So + # the MicroVM never suspends, and a state check would need + # lambda:GetMicrovm for nothing. + endpoint = fake_endpoint() + client = fake_client(endpoint=endpoint.url) + handler, invocations = handler_across_deploy( + client, {}, {"auto_suspend_on_idle": True}, two_jobs + ) + + result = run(handler, None, [("build", succeed_with(1)), ("test", succeed_with(2))]) + + assert outcome(result)["result"]["test"] == 2 + assert len(invocations) > 1 + [launch] = client.requests("run_microvm") + assert "autoSuspendIdleSeconds" not in json.loads(launch["runHookPayload"]) + assert client.requests("get_microvm") == [] + assert len(endpoint.requests) == 2 + assert launch_checkpoint(result)["resumeBeforeDelivery"] is False + + +@pytest.mark.parametrize( + ("suspend", "state_checks"), + [(True, 2), (False, 0)], + ids=["current-config-can-suspend", "current-config-cannot-suspend"], +) +def test_follows_the_current_config_for_a_launch_checkpoint_without_a_resume_mode( + run: Any, + fake_client: Any, + fake_endpoint: Any, + monkeypatch: pytest.MonkeyPatch, + suspend: bool, # noqa: FBT001 - a pytest parameter + state_checks: int, +) -> None: + # Code that predates the field checkpointed the launch without + # resumeBeforeDelivery. That code computed the mode from the current + # config on every replay. + checkpoint_launch_as(monkeypatch) + client = fake_client(endpoint=fake_endpoint().url) + config = session_config(client, auto_suspend_on_idle=suspend) + + result = run( + session_handler(config, two_jobs), + None, + [("build", succeed_with(1)), ("test", succeed_with(2))], + ) + + assert outcome(result)["result"]["test"] == 2 + assert "resumeBeforeDelivery" not in launch_checkpoint(result) + assert len(client.requests("get_microvm")) == state_checks + + +@pytest.mark.parametrize( + ("recorded", "suspend", "state_checks"), + [("false", False, 0), (1, False, 0), (0, True, 2)], + ids=["string-false", "one", "zero"], +) +def test_follows_the_current_config_for_a_resume_mode_that_is_not_a_bool( + run: Any, + fake_client: Any, + fake_endpoint: Any, + monkeypatch: pytest.MonkeyPatch, + recorded: Any, + suspend: bool, # noqa: FBT001 - a pytest parameter + state_checks: int, +) -> None: + # Only a bool is a resume mode. A truthy or falsy value of another type + # must not decide whether a job checks the MicroVM state. In Python + # True == 1, so a check by equality would read 1 as True and 0 as False. + checkpoint_launch_as(monkeypatch, resumeBeforeDelivery=recorded) + client = fake_client(endpoint=fake_endpoint().url) + config = session_config(client, auto_suspend_on_idle=suspend) + + result = run( + session_handler(config, two_jobs), + None, + [("build", succeed_with(1)), ("test", succeed_with(2))], + ) + + assert outcome(result)["result"]["test"] == 2 + checkpointed = launch_checkpoint(result)["resumeBeforeDelivery"] + assert type(checkpointed) is type(recorded) + assert checkpointed == recorded + assert len(client.requests("get_microvm")) == state_checks