diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/worker_composed_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/worker_composed_test.py index 06aaf0464..ebc397a1d 100644 --- a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/worker_composed_test.py +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/worker_composed_test.py @@ -19,6 +19,7 @@ import functools import http.client +import http.server import json import socket import threading @@ -557,3 +558,163 @@ def handler(event: Any, context: DurableContext) -> Any: assert refusals == 0 else: assert refusals >= 1 + + +class LosingProxy: + """A stand-in for the MicroVM endpoint in front of a real worker. + + It forwards each job request to the worker, and returns the worker's + status. Before it forwards a request, it runs the action for that + request's number, if there is one. It can drop the worker's answer to a + request, as a network fault does. + """ + + def __init__( + self, + worker: MicrovmWorker, + *, + lose: set[int], + before: dict[int, Callable[[], None]], + ) -> None: + self.statuses: list[int | None] = [] + self._lock = threading.Lock() + proxy = self + + class Handler(http.server.BaseHTTPRequestHandler): + def do_POST(self) -> None: # noqa: N802 - the http.server name + body = self.rfile.read(int(self.headers.get("content-length", "0"))) + with proxy._lock: + number = len(proxy.statuses) + 1 + proxy.statuses.append(None) + action = before.get(number) + if action is not None: + action() + connection = http.client.HTTPConnection( + host_of(worker), worker.port, timeout=WAIT + ) + try: + connection.request( + "POST", + self.path, + body=body, + headers={"content-type": "application/json"}, + ) + response = connection.getresponse() + response.read() + status = response.status + finally: + connection.close() + if number in lose: + self.close_connection = True + self.connection.shutdown(socket.SHUT_RDWR) + return + with proxy._lock: + proxy.statuses[number - 1] = status + self.send_response(status) + self.send_header("content-length", "0") + self.end_headers() + + def log_message(self, *args: Any) -> None: + pass + + self._server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + self.url = f"http://127.0.0.1:{self._server.server_address[1]}" + threading.Thread(target=self._server.serve_forever, daemon=True).start() + + def close(self) -> None: + self._server.shutdown() + self._server.server_close() + + +def test_resumes_the_microvm_when_a_resend_of_a_running_job_meets_a_suspend( + fake_client: Any, +) -> None: + """A resend of a running job during a suspend gets 503, not 202. + + 1. The worker accepts the job, and its 202 is lost. + 2. The request step checks the state again, and sees ``RUNNING``. + 3. Lambda sends the ``suspend`` hook, and the MicroVM is ``SUSPENDED``. + 4. The resend reaches the worker during the suspend refusal. + + A 202 at step 4 would end the delivery. Nothing would resume the + MicroVM, and the frozen job would never complete its callback. The 503 + makes the request step check the state again, see ``SUSPENDED``, and + resume the MicroVM. The job here continues only after ResumeMicrovm, as + a frozen job does. + """ + state = {"value": "RUNNING"} + lock = threading.Lock() + resumed = threading.Event() + holder: dict[Any, Any] = {} + + class StatefulMicrovms(fake_client): # type: ignore[misc, valid-type] + def get_microvm(self, **request: Any) -> dict[str, Any]: + with lock: + current = state["value"] + response = super().get_microvm(**request) + return {**response, "state": current} + + def resume_microvm(self, **request: Any) -> dict[str, Any]: + super().resume_microvm(**request) + with lock: + state["value"] = "RUNNING" + resumed.set() + post_hook(holder["worker"], "resume") + return {} + + def suspend() -> None: + post_hook(holder["worker"], "suspend") + with lock: + state["value"] = "SUSPENDED" + + runs: list[str] = [] + + def job(job_input: Any, context: MicrovmJobContext) -> Any: + runs.append(job_input) + # A frozen job continues only after the MicroVM resumes. + return "done" if resumed.wait(10) else "still frozen" + + def pipeline(vm: MicrovmSession, ctx: DurableContext) -> Any: + options = MicrovmInvokeOptions(timeout=Duration.from_minutes(1)) + return vm.invoke(ctx, "build", job_input="build", options=options) + + @durable_execution + def handler(event: Any, context: DurableContext) -> Any: + return microvm_session( + context, "pipeline", config=holder["config"], handler=pipeline + ) + + runner = DurableFunctionTestRunner(handler=handler, poll_interval=0.05) + callbacks = RunnerCallbacks(runner) + worker = start_reachable_worker( + lambda: start_microvm_worker( + handler=job, + create_client=lambda region: callbacks, # type: ignore[arg-type, return-value] + port=0, + ) + ) + holder["worker"] = worker + proxy = LosingProxy(worker, lose={1}, before={2: suspend}) + try: + client = StatefulMicrovms(launches=[run_hook_to(worker)], endpoint=proxy.url) + holder["config"] = MicrovmSessionConfig( + image_identifier="img", + execution_role_arn="role", + timeout=Duration.from_minutes(5), + auto_suspend_on_idle=True, + auto_suspend_idle_time=Duration.from_seconds(60), + client=client, + ) + arn = runner.run_async(input="null", execution_timeout=WAIT) + result = runner.wait_for_result(arn, timeout=WAIT) + assert worker.idle(WAIT) + finally: + proxy.close() + worker.close() + runner.close() + + assert result.status is InvocationStatus.SUCCEEDED, result.error + assert json.loads(result.result or "null") == "done" + assert runs == ["build"] + assert proxy.statuses == [None, 503, 202] + assert client.requests("resume_microvm") == [{"microvmIdentifier": "mvm-1"}] diff --git a/packages/aws-durable-execution-sdk-python-microvm-worker/README.md b/packages/aws-durable-execution-sdk-python-microvm-worker/README.md index 851063f90..02ffa19fe 100644 --- a/packages/aws-durable-execution-sdk-python-microvm-worker/README.md +++ b/packages/aws-durable-execution-sdk-python-microvm-worker/README.md @@ -92,7 +92,11 @@ Each job runs in its own thread. For every job, the worker: 3. Completes the callback with the handler's result, or fails it with the handler's exception. A `SystemExit` from the handler is reported as a failure too. 4. Reports a result that is over 256 KB, or that is not JSON-serializable, as a failure. So the durable function learns why at once, instead of at its timeout. -A second delivery of a job is answered like the first and ignored, while the job runs and after it ended. The durable function can deliver the same job again: it retries a delivery whose answer it did not receive, and its request step runs again on replay when its result was not checkpointed. The worker remembers the callback IDs of the last 1,000 finished jobs. When a job ends and the worker already holds 1,000 IDs, it removes the oldest one. So a job runs once, unless at least 1,000 other jobs ended in between, or the worker process restarted. +A second delivery of a job is answered like the first and ignored, while the job runs and after it ended. The durable function can deliver the same job again: it retries a delivery whose answer it did not receive, and its request step runs again on replay when its result was not checkpointed. The worker remembers the callback IDs of the last 1,000 finished jobs. When a job ends and the worker already holds 1,000 IDs, it removes the oldest one. So a job runs once, unless at least 1,000 other jobs ended in between, or the worker process restarted. While the worker refuses new jobs after a suspend or after the `terminate` hook, a second delivery gets one of three answers: + +1. A second delivery of a finished job gets HTTP 202 in every state. It starts nothing. A 503 would make the durable function resume the MicroVM only to get this answer. +2. A second delivery of a known job gets HTTP 202 after the `terminate` hook. That hook fails a running job's callback. So the durable function learns the outcome from the callback. +3. A second delivery of a running job gets HTTP 503 during a suspend refusal, as a new job does. The running job can freeze with the MicroVM. A 202 would end the delivery, and nothing would resume the MicroVM. The job would then wait for its heartbeat timeout or its timeout. The 503 makes the durable function check the state, resume the MicroVM, and deliver the job again. That delivery gets HTTP 202, and the job completes once. The worker creates a Lambda client for each job, after the job arrives, not at image build. A client created before the snapshot would carry build-time state into every MicroVM. A client kept across jobs would keep connections that a suspend of the MicroVM can leave dead. The worker closes each job's default client when the job ends. It never closes a client from `create_client`. @@ -106,19 +110,19 @@ A session's run hook payload can set `autoSuspendIdleSeconds`. The session sets 1. The idle time starts when the `run` hook arrives without a job, when a job ends, and when the `resume` lifecycle hook arrives, if the image enables it. 2. A job that starts cancels the idle time. So the worker never suspends the MicroVM during a job. -3. When the idle time ends, the worker calls `SuspendMicrovm` with the MicroVM ID from the `run` hook, never one from a job request. The call returned in under 100 milliseconds in testing, before the MicroVM froze. No attempt of the call starts after 10 seconds, because a `before-send` handler on the client stops botocore's retries at that deadline. botocore cannot cancel an attempt in flight, and it opens the connection after `before-send`. So an attempt that started before the deadline can still reach the service after it. The worker waits for that attempt to end, and refuses jobs until then. The client's connect and read timeouts bound the attempt: 5 and 10 seconds for the default client. A client from `create_microvms_client` should use similar timeouts. botocore's own defaults are 60 seconds each, so with them the refusal can last minutes. The durable function resumes the MicroVM before the next job, and the process continues with its memory and files. -4. From the moment the idle time ends, the worker answers every job with HTTP 503 and does not start it. A job accepted then would freeze with the MicroVM, and nothing would resume it. After a 503, the durable function checks the state, resumes the MicroVM, and delivers the job again. -5. The refusal ends when the `resume` lifecycle hook arrives, or 30 seconds after the call returned, if no hook arrives. The idle time then starts again. At most one call runs at a time. A `resume` hook that arrives during the call ends the refusal at once, and the next idle time starts when the call ends. So enable the `resume` hook in the image. Without it, the first job after a resume can wait a little over 30 seconds: the 30 seconds, plus up to 4 seconds until the durable function's next delivery attempt. A test measured 32 seconds. +3. When the idle time ends, the worker calls `SuspendMicrovm` with the MicroVM ID from the `run` hook, never one from a job request. The call returned in under 100 milliseconds in testing, before the MicroVM froze. The default client makes one attempt. An attempt can suspend the MicroVM and lose its answer in the freeze. A retry would then run after the MicroVM resumed. The `resume` hook has ended the refusal of jobs by then. So the retry could suspend the MicroVM with the next job inside. With one attempt, a 5xx answer or a dropped connection on the first attempt of the default client counts as unknown, as in item 8. botocore's retry would often have succeeded. So the worker refuses jobs for up to 30 seconds on a MicroVM that did not suspend. No `resume` hook ends that refusal, because nothing suspended. This delay is the accepted cost of one attempt. No attempt of the call starts after 10 seconds, because a `before-send` handler on the client stops botocore's retries at that deadline. The same handler starts no attempt after a `resume` hook. This covers a client from `create_microvms_client` that retries. botocore cannot cancel an attempt in flight, and it opens the connection after `before-send`. So an attempt that started before the deadline can still reach the service after it. The worker waits for that attempt to end, and refuses jobs until then. The client's connect and read timeouts bound the attempt: 5 and 10 seconds for the default client. A client from `create_microvms_client` should use similar timeouts. botocore's own defaults are 60 seconds each, so with them the refusal can last minutes. The durable function resumes the MicroVM before the next job, and the process continues with its memory and files. +4. From the moment the idle time ends, the worker answers every new job with HTTP 503 and does not start it. A job accepted then would freeze with the MicroVM, and nothing would resume it. After a 503, the durable function checks the state, resumes the MicroVM, and delivers the job again. +5. The refusal ends when the `resume` lifecycle hook arrives, or 30 seconds after the call returned, if no hook arrives. The idle time then starts again. At most one call runs at a time. A `resume` hook that arrives during the call ends the refusal at once, and the next idle time starts when the call ends. The call then starts no further attempt, and the worker logs `stopping the suspend call: the MicroVM resumed during it` at info level. A second `resume` hook during the same call logs nothing. The JavaScript worker cancels the attempt in flight. This worker can only stop the next attempt, because botocore cannot cancel an attempt in flight. So enable the `resume` hook in the image. Without it, the first job after a resume can wait a little over 30 seconds: the 30 seconds, plus up to 4 seconds until the durable function's next delivery attempt. A test measured 32 seconds. 6. A call that the service rejected on its first attempt, such as `AccessDeniedException` for a missing permission, is logged as a warning, and the MicroVM keeps running. Missing credentials count as a rejection. A later call in the same MicroVM most likely finds no credentials either. The worker accepts jobs again at once and does not retry the call. The next job that ends starts the idle time again. The exception is a `resume` hook that arrived during the call. The idle time then starts when the call ends. So the worker retries the call one idle time after the call ends, if no job runs and no `suspend` hook refusal is in effect when the call ends. A second exception is a `suspend` hook that arrived during the call and whose refusal is still in effect when the call ends. The MicroVM is then suspending anyway. So the rejection does not end the hook's refusal. That refusal ends on the `resume` hook, or 30 seconds after the last `suspend` hook. The idle time then starts again. -7. A call that the service refused for now on its first attempt is logged as a warning too, and the worker accepts jobs again at once. This covers expired credentials and a signature error, with or without a clock offset, whatever the HTTP status. The service answers these before it acts. botocore retries none of these errors, so they reach the worker on the first attempt. The idle time then starts again, so the worker tries the call again. That idle time lasts at least 30 seconds, also when `autoSuspendIdleSeconds` is shorter. The same error can repeat on every call, for example with a wrong secret key or an expired session token from environment variables. So while no job ends and no lifecycle hook arrives, the worker makes at most one call per 30 seconds, and logs one warning per call. A job that ends after the refusal, or a `resume` hook after it, starts the normal idle time instead. A `resume` hook or a job that ends during the call does not shorten the idle time after the refusal, because the refused outcome arrives after it. A `suspend` hook whose refusal is still in effect when the call ends is the exception, as in item 6. Its refusal stays, and the idle time starts when it ends. The default client retries a throttle, so a throttle from it usually ends as an unknown outcome, as in item 8. A throttle counts as refused for now only from a client that does not retry it, such as a client from `create_microvms_client` with one attempt. The JavaScript worker classifies these errors the same way. Its default client retries a clock-skew error. So there, a clock-skew error usually ends as an unknown outcome. -8. A call whose outcome is unknown is handled like a successful call, because the service may have suspended the MicroVM anyway. This covers a timeout, a conflict, a server error, and a dropped connection. An error after botocore retried an earlier attempt also counts as unknown, because the earlier attempt may have suspended the MicroVM. The default client uses botocore's standard retry mode with 3 attempts, and that mode retries a throttle. So a throttle usually ends here. The refusal continues until the `resume` hook or the 30-second limit. The idle time then starts again, so the worker tries the call again. A `resume` hook during the call, or the end of a `suspend` hook's 30 seconds during the call, has already ended the refusal. The worker then accepts jobs at once, and tries the call again one idle time after the call and every job end. -9. The worker logs one warning for each failed call. The message names the outcome. The fields are, in order, `microvmId`, `error`, `outcome`, `suspendHookArrived`, `terminating`, `closed`, and `refusingJobs`. `microvmId` is the MicroVM ID of the call. `error` holds the name and the message of the error. `outcome` is `rejected`, `refusedForNow`, or `unknown`. `suspendHookArrived` states whether a `suspend` hook arrived since the idle time ended. `terminating` states whether a `terminate` hook came before the worker acted on the outcome. `closed` states whether `worker.close()` came before the worker acted on the outcome. `refusingJobs` states whether the worker answers jobs with HTTP 503 after it has acted on the outcome. +7. A call that the service refused for now on its first attempt is logged as a warning too, and the worker accepts jobs again at once. This covers expired credentials and a signature error, with or without a clock offset, whatever the HTTP status. The service answers these before it acts. botocore retries none of these errors, so they reach the worker on the first attempt. The idle time then starts again, so the worker tries the call again. That idle time lasts at least 30 seconds, also when `autoSuspendIdleSeconds` is shorter. The same error can repeat on every call, for example with a wrong secret key or an expired session token from environment variables. So while no job ends and no lifecycle hook arrives, the worker makes at most one call per 30 seconds, and logs one warning per call. A job that ends after the refusal, or a `resume` hook after it, starts the normal idle time instead. A `resume` hook or a job that ends during the call does not shorten the idle time after the refusal, because the refused outcome arrives after it. A `suspend` hook whose refusal is still in effect when the call ends is the exception, as in item 6. Its refusal stays, and the idle time starts when it ends. The default client makes one attempt. So a throttle from it counts as refused for now. A client from `create_microvms_client` that retries a throttle turns it into an unknown outcome, as in item 8. The JavaScript worker classifies these errors the same way. +8. A call whose outcome is unknown is handled like a successful call, because the service may have suspended the MicroVM anyway. This covers a timeout, a conflict, a server error, and a dropped connection. An error after botocore retried an earlier attempt also counts as unknown, because the earlier attempt may have suspended the MicroVM. Only a client from `create_microvms_client` that retries can raise such an error. A call that the `before-send` handler stopped after a `resume` hook counts as unknown too. The handler can stop the first attempt, and then nothing was sent. An earlier attempt, if there was one, may have suspended the MicroVM. That `resume` hook has already ended the refusal. So the worker accepts jobs at once, and starts no new 30-second limit for that call. The refusal continues until the `resume` hook or the 30-second limit. The idle time then starts again, so the worker tries the call again. A `resume` hook during the call, or the end of a `suspend` hook's 30 seconds during the call, has already ended the refusal. The worker then accepts jobs at once, and tries the call again one idle time after the call and every job end. +9. The worker logs one warning for each failed call. The message names the outcome. The fields are, in order, `microvmId`, `error`, `outcome`, `suspendHookArrived`, `terminating`, `closed`, and `refusingJobs`. `microvmId` is the MicroVM ID of the call. `error` holds the name and the message of the error. `outcome` is `rejected`, `refusedForNow`, or `unknown`. `suspendHookArrived` states whether a `suspend` hook arrived since the idle time ended. `terminating` states whether a `terminate` hook came before the worker acted on the outcome. `closed` states whether `worker.close()` came before the worker acted on the outcome. `refusingJobs` states whether the worker answers jobs with HTTP 503 after it has acted on the outcome. The worker also logs two info lines about the call. It logs `suspending the idle MicroVM` before each call, with `microvmId`, `idleSeconds`, and `waitedSeconds`. It logs `stopping the suspend call: the MicroVM resumed during it` for the first `resume` hook during a call, with `microvmId`. The JavaScript worker logs the same stop line. `worker.close()` stops the idle time. A closed worker never suspends its MicroVM again. A refusal that is already running still ends on the `resume` hook or after 30 seconds. The worker counts only jobs. It cannot see background threads or processes, or inbound traffic that is not a job. A session with such work leaves `auto_suspend_on_idle` off, and the payload then has no `autoSuspendIdleSeconds`. -The worker creates a `lambda-microvms` client for each suspend, and closes it after the call. Pass `create_microvms_client` to provide the client. The worker never closes a client that it returns. A client that retries must report its retries in `ResponseMetadata.RetryAttempts` of the error that it raises, as botocore does. The worker otherwise counts a retried call as a first attempt. An earlier attempt of that call may have suspended the MicroVM. So a retried throttle would count as refused for now, and the worker would accept a job that then freezes with the MicroVM. +The worker creates a `lambda-microvms` client for each suspend, and closes it after the call. Pass `create_microvms_client` to provide the client. The worker never closes a client that it returns. Give the client one attempt, as the default client has. The worker starts no attempt after a `resume` hook. A retry that starts before the `resume` hook arrives can still suspend the MicroVM after it resumed. A client that retries must report its retries in `ResponseMetadata.RetryAttempts` of the error that it raises, as botocore does. The worker otherwise counts a retried call as a first attempt. An earlier attempt of that call may have suspended the MicroVM. So a retried throttle would count as refused for now, and the worker would accept a job that then freezes with the MicroVM. ## Suspend and terminate hooks @@ -127,7 +131,7 @@ Lambda waits for the answer to a `suspend` or `terminate` hook, up to the hook's On the `suspend` hook the worker: 1. Answers HTTP 200 at once. A suspend hook that ran past its timeout terminated the MicroVM instead of suspending it. -2. Refuses every job with HTTP 503, as after its own suspend. A job accepted now would freeze with the MicroVM. The refusal ends when the `resume` hook arrives, or after 30 seconds. +2. Refuses every new job, and every second delivery of a running job, with HTTP 503, as after its own suspend. A job accepted now would freeze with the MicroVM. The refusal ends when the `resume` hook arrives, or after 30 seconds. 3. Leaves running jobs alone. They freeze with the MicroVM, and continue after a resume. 4. Defers a job from a `run` hook whose body was still arriving. The durable function cannot deliver a `run` hook job again, so the worker does not refuse it. It starts the job when the refusal ends. @@ -139,7 +143,7 @@ On the `terminate` hook the worker: 2. Fails each running job's callback with `MicrovmTerminatedError`. The worker reports `MicrovmTerminatedError` as the callback's error type. The `microvm` operation then fails with a `MicrovmJobFailedError` whose message contains `(MicrovmTerminatedError)`. The worker does not report the handler's outcome when it returns later. 3. Skips a job whose callback a heartbeat already found gone. A report would fail the same way. 4. Answers HTTP 200 when the reports end, or after 5 seconds. A report took 20 to 1,700 milliseconds in testing. The answer also waits for a job that is reporting its own outcome, for a `run` hook request whose body is still arriving, and for every callback failure report in flight, such as the report for an invalid request. Work that begins during the wait is included. A second `terminate` hook that arrives during the wait waits for the same work. Lambda can end the MicroVM as soon as it gets the answer, so a report sent after it can be lost. -5. Refuses every later job with HTTP 503, and never suspends the MicroVM again. A job from a `run` hook whose body was still arriving is not started. Its callback fails with `MicrovmTerminatedError`, as does a job deferred by an earlier `suspend` hook. +5. Refuses every later new job with HTTP 503, and never suspends the MicroVM again. A second delivery of a known job gets HTTP 202. That job has ended, or this hook fails its callback. So the durable function learns the outcome from the callback, and a 503 would only make it deliver the job again. A job whose callback a heartbeat already found gone gets HTTP 202 too. That callback is gone, so the durable function no longer waits for it. A job from a `run` hook whose body was still arriving is not started. Its callback fails with `MicrovmTerminatedError`, as does a job deferred by an earlier `suspend` hook. Lambda called the hook after `TerminateMicrovm` on a running MicroVM, and at the end of `maximumDurationInSeconds`. It did not call the hook when it terminated a suspended MicroVM, after `TerminateMicrovm` or at the end of the idle policy's `suspendedDurationSeconds`. The process is frozen then. A job of such a MicroVM fails at its heartbeat timeout or its timeout. diff --git a/packages/aws-durable-execution-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/self_suspend.py b/packages/aws-durable-execution-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/self_suspend.py index 21ecaf6b0..948bfe422 100644 --- a/packages/aws-durable-execution-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/self_suspend.py +++ b/packages/aws-durable-execution-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/self_suspend.py @@ -1,12 +1,22 @@ # SPDX-FileCopyrightText: 2026-present Amazon.com, Inc. or its affiliates. # # SPDX-License-Identifier: Apache-2.0 -"""The worker's own SuspendMicrovm call, for a session that suspends when idle.""" +"""The worker's own SuspendMicrovm call, for a session that suspends when idle. + +The call can suspend the MicroVM and lose its answer in the freeze. A retry +of the call then runs after the MicroVM resumed. The ``resume`` hook has +already ended the worker's refusal of jobs at that point. So the retry would +suspend the MicroVM with the next job inside it. So the worker stops the +call's attempts after a ``resume`` hook, and its default client makes one +attempt. After a failed call, the worker can call again after a later idle +time, with a new refusal of jobs. +""" from __future__ import annotations import time import uuid +from collections.abc import Callable from enum import Enum from typing import TYPE_CHECKING, Any @@ -53,10 +63,13 @@ _CONNECT_TIMEOUT_SECONDS = 5 -_CLIENT_ATTEMPTS = 3 +_CLIENT_ATTEMPTS = 1 """The attempts of the default client, the first attempt included. -The JavaScript worker's default client makes 3 attempts too. +An attempt can suspend the MicroVM and lose its answer in the freeze. A +retry then runs after the MicroVM resumed, and can suspend it with a job +inside. So the default client makes one attempt. After a failed call, the +worker can call again after a later idle time, with a new refusal of jobs. """ _BEFORE_SEND = "before-send.lambda-microvms.SuspendMicrovm" @@ -77,12 +90,10 @@ class SuspendOutcome(Enum): """The service did nothing on the call's first and only attempt, and a later call can succeed. - Examples are expired credentials and a signature error, with or without - a clock offset. botocore retries none of these, so they keep this - outcome. The default client retries a throttle. So a throttle usually - ends as :attr:`UNKNOWN`. A throttle has this outcome only when it - reaches the worker without a retry, for example from a caller's client - that does not retry it.""" + Examples are expired credentials, a signature error with or without a + clock offset, and a throttle. The default client makes one attempt, so + these errors keep this outcome. A caller's client that retries a + throttle turns it into :attr:`UNKNOWN`.""" UNKNOWN = "unknown" """The service may have suspended the MicroVM.""" @@ -95,17 +106,30 @@ class SuspendDeadlineError(TimeoutError): """ +class SuspendResumedError(Exception): + """A ``resume`` hook arrived before an attempt of the SuspendMicrovm call + could start. + + The ``resume`` hook ends the worker's refusal of jobs. So a job can run + in the MicroVM from then on. The attempt would suspend the MicroVM with + that job inside. So the call stops before the attempt. botocore does not + retry this error. + """ + + def create_microvms_client(region: str) -> LambdaMicrovmsClient: """Create the default ``lambda-microvms`` client for one call. Each call gets its own session, because boto3's default session is not safe to use from several threads at once. - The client uses botocore's standard retry mode with 3 attempts. That - mode retries a throttle error code, such as ``ThrottlingException``, a - 5xx answer, and a connection error. It retries no credential or - signature error, also with a clock offset. Each ``ClientError`` reports the - retries in ``ResponseMetadata.RetryAttempts``. + The client makes one attempt. An attempt can suspend the MicroVM and + lose its answer in the freeze. A retry would then run after the MicroVM + resumed, and could suspend it with the next job inside. The worker calls + again after a later idle time instead, with a new refusal of jobs. + :func:`suspend_microvm` also stops the attempts of a caller's client + after the MicroVM resumed. One attempt is still the recommended setting + for such a client. """ return boto3.session.Session().client( "lambda-microvms", @@ -123,10 +147,13 @@ def create_microvms_client(region: str) -> LambdaMicrovmsClient: def suspend_microvm( - client: LambdaMicrovmsClient, microvm_id: str, deadline_seconds: float + client: LambdaMicrovmsClient, + microvm_id: str, + deadline_seconds: float, + resumed: Callable[[], bool] | None = None, ) -> None: - """Call SuspendMicrovm in this thread, and start no attempt after - ``deadline_seconds``. + """Call SuspendMicrovm in this thread. Start no attempt after + ``deadline_seconds``, and none after ``resumed()`` returns ``True``. botocore has no deadline for a whole call. Its read timeout limits one socket read, and it covers neither credential resolution nor its own @@ -136,6 +163,18 @@ def suspend_microvm( botocore went on retrying, and an attempt started 22.5 seconds after the first one. + The same handler stops an attempt after the MicroVM resumed: + + 1. An attempt can suspend the MicroVM and lose its answer in the freeze. + 2. botocore then retries the attempt after the MicroVM resumed. + 3. The ``resume`` hook has ended the worker's refusal of jobs by then. + 4. So the retry would suspend the MicroVM with a new job inside. + + So the handler raises :class:`SuspendResumedError` when ``resumed()`` + returns ``True``. botocore does not retry that error either. The handler + runs in this thread. So ``resumed`` must not block, and must not take a + lock that the caller of this function can hold. + botocore emits ``before-send`` before it opens the connection, and it cannot cancel an attempt in flight. So an attempt that started before the deadline can still reach the service after it. This function @@ -149,6 +188,8 @@ def suspend_microvm( Raises: SuspendDeadlineError: When an attempt would start after the deadline. + SuspendResumedError: When an attempt would start after ``resumed()`` + returned ``True``. Exception: The call's own error. """ deadline = time.monotonic() + deadline_seconds @@ -157,7 +198,10 @@ def suspend_microvm( unique_id = f"aws-durable-execution-microvm-worker-deadline-{uuid.uuid4().hex}" events = safe_get(lambda: client.meta.events) - def stop_after_deadline(**_: object) -> None: + def stop_attempt(**_: object) -> None: + if resumed is not None and resumed(): + msg = "the MicroVM resumed before this SuspendMicrovm attempt" + raise SuspendResumedError(msg) if time.monotonic() >= deadline: msg = ( f"the SuspendMicrovm deadline of {deadline_seconds} seconds has passed" @@ -165,7 +209,7 @@ def stop_after_deadline(**_: object) -> None: raise SuspendDeadlineError(msg) if events is not None: - events.register_first(_BEFORE_SEND, stop_after_deadline, unique_id=unique_id) + events.register_first(_BEFORE_SEND, stop_attempt, unique_id=unique_id) try: client.suspend_microvm(microvmIdentifier=microvm_id) finally: @@ -179,7 +223,8 @@ def classify_suspend_error(error: BaseException) -> SuspendOutcome: 1. A botocore retry inside the call means that an earlier attempt may have suspended the MicroVM before its answer was lost. So any error after a retry is :attr:`SuspendOutcome.UNKNOWN`. The default client - retries a throttle. So a throttle from it usually ends here. + makes one attempt. So only a caller's client that retries reaches + this rule. 2. Missing credentials are :attr:`SuspendOutcome.REJECTED`, as in the JavaScript worker. A later call in the same MicroVM most likely finds no credentials either. @@ -191,13 +236,17 @@ def classify_suspend_error(error: BaseException) -> SuspendOutcome: status. The service answers them before it acts, and a later call can succeed. Some services answer a throttle with HTTP 400. botocore retries no credential or signature error, also with a clock offset. - So these errors reach this rule. A throttle reaches it only from a - client that did not retry it. + So these errors reach this rule. A throttle reaches it from the + default client, and from any client that did not retry it. 6. Any other 4xx answer except 408, 409, and 429 is :attr:`SuspendOutcome.REJECTED`. A 429 is :attr:`SuspendOutcome.REFUSED_FOR_NOW`. - 7. Any other error, such as a timeout, a dropped connection, a 408, or a - server error, is :attr:`SuspendOutcome.UNKNOWN`. + 7. Any other error is :attr:`SuspendOutcome.UNKNOWN`. Examples are a + timeout, a dropped connection, a 408, a server error, + :class:`SuspendDeadlineError`, and :class:`SuspendResumedError`. The + last two stop an attempt. When they stop the first attempt, nothing + was sent. An earlier attempt, if there was one, may have suspended + the MicroVM. """ if isinstance(error, NoCredentialsError): return SuspendOutcome.REJECTED diff --git a/packages/aws-durable-execution-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/worker.py b/packages/aws-durable-execution-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/worker.py index 3e891e580..f13a6c485 100644 --- a/packages/aws-durable-execution-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/worker.py +++ b/packages/aws-durable-execution-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/worker.py @@ -284,7 +284,13 @@ class MicrovmWorkerListener: A second delivery of a callback ID is answered like the first and ignored, while that job runs and after it ended. The listener remembers - the last 1,000 finished callback IDs. + the last 1,000 finished callback IDs. A second delivery of a finished + job gets this answer in every state, because it starts nothing. A + second delivery of a known job gets it after the ``terminate`` hook too, + because that hook fails a running job's callback. A second delivery of a + running job during a suspend refusal gets HTTP 503, as a new job does. + The running job can freeze with the MicroVM. The 503 makes the durable + function check the state and resume the MicroVM. A document that does not match the contract gets HTTP 400. If the document still names a callback, the listener fails that callback. So @@ -301,9 +307,9 @@ class MicrovmWorkerListener: 3. When the idle time ends, the listener calls SuspendMicrovm with the ``run`` hook's MicroVM ID. The call returns before the MicroVM freezes. - 4. From the moment the idle time ends, the listener answers every job - with HTTP 503 and does not start it. A job accepted then would freeze - with the MicroVM. After a 503, the durable function resumes the + 4. From the moment the idle time ends, the listener answers every new + job with HTTP 503 and does not start it. A job accepted then would + freeze with the MicroVM. After a 503, the durable function resumes the MicroVM and delivers the job again. 5. The refusal ends when the ``resume`` hook arrives, or 30 seconds after the call returned. The idle time then starts again. @@ -330,14 +336,15 @@ class MicrovmWorkerListener: arrives after it. A ``suspend`` hook whose refusal is still in effect when the call ends is the exception, as in item 6. Its refusal stays, and the idle time starts when it ends. - The default client retries a throttle. So a throttle from it usually - ends as an unknown outcome, as in item 8. A throttle is refused for - now only from a caller's client that does not retry it. + The default client makes one attempt. So a throttle from it is + refused for now. A caller's client that retries a throttle turns it + into an unknown outcome, as in item 8. 8. A call whose outcome is unknown, such as a timeout, a conflict, a server error, or an error after a retried attempt, is handled like a successful call. The service may have suspended the MicroVM anyway. - No attempt of the call starts after 10 seconds. An attempt already in - flight runs until it ends, and the refusal lasts at least until then. + No attempt of the call starts after 10 seconds, and none starts + after a ``resume`` hook. An attempt already in flight runs until it + ends, and the refusal lasts at least until then. A `resume` hook during the call, or the end of a `suspend` hook's 30 seconds during the call, has already ended the refusal. The worker then accepts jobs at once, and tries the call again one idle time @@ -360,7 +367,8 @@ class MicrovmWorkerListener: - ``suspend``: the listener answers HTTP 200 at once, because a suspend hook that runs past its timeout terminates the MicroVM. It then - refuses jobs with HTTP 503, as after its own suspend, until the + refuses new jobs, and second deliveries of running jobs, with HTTP + 503, as after its own suspend, until the ``resume`` hook or for 30 seconds. Running jobs freeze with the MicroVM. A ``run`` hook job that arrives during the refusal starts when the refusal ends, because the durable function cannot deliver it @@ -370,8 +378,8 @@ class MicrovmWorkerListener: :class:`MicrovmTerminatedError`. It answers HTTP 200 when the reports end, or after 5 seconds. The answer also waits for jobs that report their own outcome, for ``run`` hook requests in flight, and for the - failure reports of invalid requests. From then on it refuses jobs with - HTTP 503, and fails the callback of a ``run`` hook job instead of + failure reports of invalid requests. From then on it refuses new jobs + with HTTP 503, and fails the callback of a ``run`` hook job instead of starting it. The listener acts on these two hooks, and on ``resume``, only when the @@ -454,6 +462,19 @@ def __init__( # worker's own that the service rejected must not end a refusal that # a suspend hook started meanwhile. self._suspend_hooks = 0 + # Counts the resume hooks from Lambda. Each one ends the refusal, so + # a job can run from then on. A SuspendMicrovm attempt after it would + # freeze that job. So the call's before-send handler starts no + # attempt once this count has changed. The handler reads the count + # without the lock: an int read is atomic, and the handler then + # waits for no listener lock inside botocore. + self._resume_hooks = 0 + # The MicroVM ID of the SuspendMicrovm call in flight. It is set when + # the idle time ends, and cleared when the call returns or the first + # resume hook during the call arrives. That hook logs that it stops + # the call. So a second resume hook during the same call logs + # nothing. + self._stoppable_suspend: str | None = None # SuspendMicrovm calls in flight. idle() waits for them. At most one # runs at a time. self._suspend_calls = 0 @@ -511,7 +532,11 @@ def create( idle, once per suspend. It never closes a client that this function returns. Defaults to a client of its own per suspend, from the default credential chain. The listener - closes that client after the call. A client that retries + closes that client after the call. The listener stops the + client's attempts once a ``resume`` hook arrived. Give the + client one attempt, as the default client has. A retry + that starts before the ``resume`` hook arrives can still + suspend the MicroVM after it resumed. A client that retries must report its retries in ``ResponseMetadata.RetryAttempts`` of the error that it raises. The listener otherwise counts a retried call as a first attempt. An earlier attempt of that @@ -708,13 +733,28 @@ def _handle_resume_hook(self) -> list[tuple[MicrovmJobDocument, str]]: A resume without a job would otherwise leave the MicroVM running until the next job ends. Returns the deferred run hook jobs, - registered. The caller starts them after the answer. + registered. The caller starts them after the answer. A + SuspendMicrovm call in flight starts no attempt after this hook. + The first such hook logs that it stops the call. """ with self._locked(): + self._resume_hooks += 1 + stopped = self._stoppable_suspend + self._stoppable_suspend = None if self._suspending: - return self._end_suspending() - self._start_idle_time() - return [] + deferred = self._end_suspending() + else: + self._start_idle_time() + deferred = [] + # The caller of the listener passes the logger. So the line is + # written after the lock is released. botocore cannot cancel an + # attempt in flight. So the call stops at its next attempt. + if stopped is not None: + self._logger.info( + "stopping the suspend call: the MicroVM resumed during it", + extra={"microvmId": stopped}, + ) + return deferred def _handle_terminate_hook(self, exchange: _Exchange) -> None: """Fail the callback of every running job, then answer the hook. @@ -951,11 +991,37 @@ def _handle_job_request( # arrives. self._microvm_id = request.microvm_id microvm_id = self._microvm_id - terminating = self._terminating - suspending = self._suspending - duplicate = self._is_known(job.callback_id) - if not (terminating or suspending or duplicate): + # A known callback ID gets one of three answers: + # + # 1. An ID in _finished gets 202 in every state. Its job has + # ended, and the answer starts nothing. A 503 would make the + # durable function resume the MicroVM only to get this answer. + # 2. A known ID gets 202 while the MicroVM terminates. The + # terminate hook fails a running job's callback. So the + # durable function learns the outcome from the callback. + # 3. An ID in _jobs gets 503 during a suspend refusal, as a new + # job does. The running job can freeze with the MicroVM. A + # 202 would end the delivery, and nothing would resume the + # MicroVM. So the job would wait for its heartbeat timeout or + # its timeout. The 503 makes the durable function check the + # state, resume the MicroVM, and deliver the job again. + may_freeze = ( + job.callback_id in self._jobs + and self._suspending + and not self._terminating + ) + duplicate = self._is_known(job.callback_id) and not may_freeze + terminating = not duplicate and self._terminating + suspending = not duplicate and self._suspending + if not (duplicate or terminating or suspending): self._register_job(job.callback_id, request.region) + if duplicate: + exchange.respond(202, {}) + self._logger.warning( + "duplicate job ignored", + extra={"microvmId": microvm_id, "callbackId": job.callback_id}, + ) + return if terminating or suspending: # Terminating: the MicroVM is about to end, and the job could not # finish. Suspending: the MicroVM is about to freeze. The durable @@ -967,13 +1033,6 @@ def _handle_job_request( ) exchange.respond(503, {"error": f"the MicroVM is {state}"}) return - if duplicate: - exchange.respond(202, {}) - self._logger.warning( - "duplicate job ignored", - extra={"microvmId": microvm_id, "callbackId": job.callback_id}, - ) - return # The thread starts before the answer. A thread that cannot start # raises, and the caller gets HTTP 500, which the durable function # retries. After a 202, the durable function does not deliver the job @@ -1438,10 +1497,14 @@ def _idle_time_ended(self, generation: int, waited_seconds: float) -> None: # thread starts, while the client is created, or during the call. # Each path that ends the refusal compares with this count. hooks_before = self._suspend_hooks + # A resume hook from here on ends the refusal. The call then + # starts no further attempt. + resumes_before = self._resume_hooks + self._stoppable_suspend = auto_suspend.microvm_id try: threading.Thread( target=self._suspend_self, - args=(auto_suspend, hooks_before, waited_seconds), + args=(auto_suspend, hooks_before, resumes_before, waited_seconds), name="microvm-suspend", daemon=True, ).start() @@ -1458,10 +1521,16 @@ def _idle_time_ended(self, generation: int, waited_seconds: float) -> None: self._launch_deferred(deferred) def _suspend_self( - self, auto_suspend: _AutoSuspend, hooks_before: int, waited_seconds: float + self, + auto_suspend: _AutoSuspend, + hooks_before: int, + resumes_before: int, + waited_seconds: float, ) -> None: try: - self._call_suspend(auto_suspend, hooks_before, waited_seconds) + self._call_suspend( + auto_suspend, hooks_before, resumes_before, waited_seconds + ) finally: with self._locked(): self._suspend_call_ended() @@ -1484,13 +1553,19 @@ def _suspend_call_ended(self) -> None: a resume hook, a job, or a refusal for now asked for during it. The caller holds the lock through :meth:`_locked`.""" self._suspend_calls -= 1 + # The thread or the client could fail before the call was sent. + self._stoppable_suspend = None if self._idle_after_call: self._idle_after_call = False self._start_idle_time(after_refusal=self._idle_after_call_floor) self._changed.notify_all() def _call_suspend( - self, auto_suspend: _AutoSuspend, hooks_before: int, waited_seconds: float + self, + auto_suspend: _AutoSuspend, + hooks_before: int, + resumes_before: int, + waited_seconds: float, ) -> None: target = auto_suspend.microvm_id fields = {"microvmId": target} @@ -1530,7 +1605,7 @@ def _call_suspend( # close(). unknown_described: object | None = None try: - suspend_microvm(client, target, SUSPEND_CALL_TIMEOUT_SECONDS) + self._send_suspend(client, target, resumes_before) except Exception as error: # noqa: BLE001 outcome = classify_suspend_error(error) # describe() calls str(error). A client from @@ -1544,7 +1619,12 @@ def _call_suspend( # A timeout, a conflict, a server error, or an error after a # retried attempt does not show whether the service suspended # the MicroVM. So the refusal continues, as after a successful - # call. The block after the client is closed acts on it. + # call. The block after the client is closed acts on it. A + # resume hook can stop any attempt, the first one included. An + # earlier attempt, if there was one, may have suspended the + # MicroVM. That resume hook has already ended the refusal. So + # that block starts a grace period only for a suspend hook + # that came after the resume. unknown_described = described else: deferred = [] @@ -1592,6 +1672,27 @@ def _call_suspend( ) self._logger.warning(message, extra=extra) + def _send_suspend( + self, client: LambdaMicrovmsClient, target: str, resumes_before: int + ) -> None: + """Make the SuspendMicrovm call. Start no attempt after a resume hook + that arrived since the idle time ended.""" + try: + # The before-send handler calls resumed() inside botocore, in + # this thread. It reads one int without the lock. So it runs no + # code of the caller, and waits for no listener lock. + suspend_microvm( + client, + target, + SUSPEND_CALL_TIMEOUT_SECONDS, + resumed=lambda: self._resume_hooks != resumes_before, + ) + finally: + # The call has returned. A resume hook from now on stops nothing, + # so it logs no stop. + with self._lock: + self._stoppable_suspend = None + def _failed_call_log( self, outcome: SuspendOutcome, diff --git a/packages/aws-durable-execution-sdk-python-microvm-worker/tests/lifecycle_hooks_test.py b/packages/aws-durable-execution-sdk-python-microvm-worker/tests/lifecycle_hooks_test.py index 01463abbf..94d7c3c7b 100644 --- a/packages/aws-durable-execution-sdk-python-microvm-worker/tests/lifecycle_hooks_test.py +++ b/packages/aws-durable-execution-sdk-python-microvm-worker/tests/lifecycle_hooks_test.py @@ -20,8 +20,12 @@ from dataclasses import dataclass, field from typing import Any +import boto3 import pytest -from botocore.exceptions import ClientError, NoCredentialsError +from botocore.awsrequest import AWSResponse, HTTPHeaders +from botocore.config import Config +from botocore.exceptions import ClientError, ConnectionClosedError, NoCredentialsError +from botocore.retries.standard import ExponentialBackoff from aws_durable_execution_sdk_python_microvm_worker import ( HOOK_PATH_PREFIX, @@ -915,6 +919,286 @@ def test_accepts_jobs_at_once_when_the_resume_hook_arrives_during_the_call( assert eventually(lambda: sorted(h.jobs.started()) == ["cb-1", "cb-2"]) +class LostAnswerService: + """Stands in for the Lambda MicroVMs service in a botocore client. + + A ``before-send`` handler answers each attempt instead of HTTP. The + first attempt "suspends" the MicroVM. Its answer is lost: the attempt + waits until :meth:`thaw`, and then fails with a closed connection, as + after a freeze. Each later attempt also "suspends" the MicroVM, answers + 200, and records the jobs that it would freeze. + """ + + def __init__(self, listener: Callable[[], MicrovmWorkerListener]) -> None: + self._listener = listener + self._thawed = threading.Event() + self._lock = threading.Lock() + self.attempts = 0 + self.frozen_jobs: list[str] = [] + + def install(self, client: Any) -> Any: + client.meta.events.register( + "before-send.lambda-microvms.SuspendMicrovm", self._send + ) + return client + + def thaw(self) -> None: + self._thawed.set() + + def _send(self, **_: Any) -> AWSResponse: + with self._lock: + self.attempts += 1 + first = self.attempts == 1 + if first: + self._thawed.wait(WAIT) + raise ConnectionClosedError(endpoint_url="https://example.com") + with self._lock: + self.frozen_jobs.extend(self._listener()._jobs) # noqa: SLF001 + headers = HTTPHeaders() + headers["Content-Type"] = "application/json" + return AWSResponse("https://example.com", 200, headers, _Body(b"{}")) + + +class _Body: + def __init__(self, data: bytes) -> None: + self._data = data + + def stream(self, **_: Any) -> Any: + yield self._data + + +@pytest.fixture +def credentials(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("AWS_PROFILE", raising=False) + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "test") + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "test") + + +@pytest.mark.parametrize("client_kind", ["default", "retrying"]) +def test_sends_no_suspend_attempt_after_the_resume_hook( + start: Any, + logger: Any, + credentials: None, + monkeypatch: pytest.MonkeyPatch, + client_kind: str, +) -> None: + # 1. The first SuspendMicrovm attempt suspends the MicroVM. + # 2. Its answer is lost in the freeze. + # 3. The resume hook ends the refusal, and the next job starts. + # 4. botocore would then retry the closed connection. The retry would + # suspend the MicroVM with that job inside. + # The default client makes one attempt. A caller's client can retry, + # and the before-send handler stops its retry after the resume hook. + monkeypatch.setattr(ExponentialBackoff, "delay_amount", lambda self, context: 0) + harness: list[Harness] = [] + service = LostAnswerService(lambda: harness[0].listener) + if client_kind == "default": + default = worker_module.create_microvms_client + monkeypatch.setattr( + worker_module, + "create_microvms_client", + lambda region: service.install(default(region)), + ) + h = start(create_microvms_client=None) + else: + retrying = service.install( + boto3.session.Session().client( + "lambda-microvms", + region_name="us-east-1", + config=Config(retries={"mode": "standard", "total_max_attempts": 3}), + ) + ) + h = start(create_microvms_client=lambda region: retrying) + harness.append(h) + + h.run_hook(IDLE) + h.timers.advance(IDLE) + assert eventually(lambda: service.attempts == 1) + assert h.job("cb-0") == 503 + assert h.hook("resume") == 200 + assert h.job("cb-1") == 202 + assert eventually(lambda: h.started("cb-1")) + service.thaw() + h.settle() + + assert service.attempts == 1 + assert service.frozen_jobs == [] + # The resume hook has ended the refusal. The failed call starts no new + # refusal and no grace period. + assert_one_warning(logger, "unknown", refusing_jobs=False) + assert h.listener._cancel_grace_timer is None # noqa: SLF001 + assert h.job("cb-2") == 202 + h.finish_job("cb-1") + h.finish_job("cb-2") + assert eventually( + lambda: {c["callbackId"] for c in h.lambda_client.recorded()} + == {"cb-1", "cb-2"} + ) + # The next idle time makes a new call, which the resume hook above does + # not stop. + h.timers.advance(IDLE) + h.settle() + assert service.attempts == 2 + assert service.frozen_jobs == [] + + +STOPPED = "stopping the suspend call: the MicroVM resumed during it" + + +def stop_lines(logger: Any) -> list[dict[str, Any]]: + """The fields of each info line that a resume hook logged when it + stopped a SuspendMicrovm call.""" + return [d for level, m, d in logger.lines if level == "info" and m == STOPPED] + + +class AnsweringService: + """Stands in for the Lambda MicroVMs service in a botocore client. + + A ``before-send`` handler counts each attempt and answers 200 instead + of HTTP. The worker's own handler runs first. So an attempt that it + stops is not counted. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self.attempts = 0 + + def install(self, client: Any) -> Any: + client.meta.events.register( + "before-send.lambda-microvms.SuspendMicrovm", self._send + ) + return client + + def _send(self, **_: Any) -> AWSResponse: + with self._lock: + self.attempts += 1 + headers = HTTPHeaders() + headers["Content-Type"] = "application/json" + return AWSResponse("https://example.com", 200, headers, _Body(b"{}")) + + +class SlowClientFactory: + """A ``create_microvms_client`` that blocks until :meth:`release`. + + It then returns the default client, with ``service`` installed. So a + test can send hooks after the idle time ended and before the first + attempt. + """ + + def __init__(self, service: AnsweringService) -> None: + self.service = service + self.creating = threading.Event() + self._release = threading.Event() + + def __call__(self, region: str) -> Any: + self.creating.set() + self._release.wait(WAIT) + return self.service.install(worker_module.create_microvms_client(region)) + + def release(self) -> None: + self._release.set() + + +def test_starts_no_attempt_after_a_resume_hook_while_the_client_is_created( + start: Any, logger: Any, credentials: None +) -> None: + # The idle time has ended, and the worker still creates the client. A + # resume hook now ends the refusal. A job can then start, and an attempt + # would freeze it. So the call sends nothing. + service = AnsweringService() + factory = SlowClientFactory(service) + h = start(create_microvms_client=factory) + h.run_hook(IDLE) + h.timers.advance(IDLE) + assert factory.creating.wait(WAIT) + assert h.job("cb-0") == 503 + assert h.hook("resume") == 200 + factory.release() + h.settle() + + assert service.attempts == 0 + data = assert_one_warning(logger, "unknown", refusing_jobs=False) + assert data["error"]["name"] == "SuspendResumedError" + assert h.listener._cancel_grace_timer is None # noqa: SLF001 + assert stop_lines(logger) == [{"microvmId": "mvm-1"}] + # The resume hook asked for the idle time during the call. So the idle + # time starts when the call ends, and its call is not stopped. + h.advance(IDLE - 0.001) + assert service.attempts == 0 + h.advance(0.001) + assert service.attempts == 1 + assert stop_lines(logger) == [{"microvmId": "mvm-1"}] + + +@pytest.mark.parametrize("end", ["resume", "grace"]) +def test_keeps_a_suspend_hook_refusal_that_follows_a_resume_that_stopped_the_call( + start: Any, logger: Any, credentials: None, end: str +) -> None: + # The resume hook ends the refusal and stops the call. A suspend hook + # then starts a refusal of its own, before the stopped call returns. + # Lambda is freezing the MicroVM, so the end of the call keeps that + # refusal. + service = AnsweringService() + factory = SlowClientFactory(service) + h = start(create_microvms_client=factory) + h.run_hook(IDLE) + h.timers.advance(IDLE) + assert factory.creating.wait(WAIT) + assert h.hook("resume") == 200 + assert h.hook("suspend") == 200 + factory.release() + h.settle() + + assert service.attempts == 0 + assert_one_warning(logger, "unknown", refusing_jobs=True, suspend_hook_arrived=True) + assert h.job("cb-1") == 503 + if end == "resume": + assert h.hook("resume") == 200 + else: + h.advance(GRACE - 0.001) + assert h.job("cb-1") == 503 + h.advance(0.001) + assert h.job("cb-1") == 202 + assert eventually(lambda: h.started("cb-1")) + # The call had returned before the second resume hook. So that hook + # stopped nothing, and logged no stop. + assert stop_lines(logger) == [{"microvmId": "mvm-1"}] + + +def test_a_second_resume_hook_during_the_call_stops_it_too_and_logs_once( + start: Any, logger: Any, credentials: None, monkeypatch: pytest.MonkeyPatch +) -> None: + # The first attempt is in flight when both resume hooks arrive. Its + # answer is lost. botocore would retry it, and the retry would freeze + # the next job. The second resume hook must not undo the stop. + monkeypatch.setattr(ExponentialBackoff, "delay_amount", lambda self, context: 0) + harness: list[Harness] = [] + service = LostAnswerService(lambda: harness[0].listener) + retrying = service.install( + boto3.session.Session().client( + "lambda-microvms", + region_name="us-east-1", + config=Config(retries={"mode": "standard", "total_max_attempts": 3}), + ) + ) + h = start(create_microvms_client=lambda region: retrying) + harness.append(h) + h.run_hook(IDLE) + h.timers.advance(IDLE) + assert eventually(lambda: service.attempts == 1) + assert h.hook("resume") == 200 + assert h.hook("resume") == 200 + service.thaw() + h.settle() + + assert service.attempts == 1 + assert_one_warning(logger, "unknown", refusing_jobs=False) + assert h.listener._cancel_grace_timer is None # noqa: SLF001 + assert stop_lines(logger) == [{"microvmId": "mvm-1"}] + assert h.job("cb-1") == 202 + assert eventually(lambda: h.started("cb-1")) + + def test_logs_a_rejected_suspend_accepts_jobs_at_once_and_does_not_retry( start: Any, logger: Any ) -> None: @@ -1236,6 +1520,30 @@ def test_a_job_that_ends_before_a_refused_call_has_ended_uses_the_normal_idle_ti assert client.suspended == ["mvm-1", "mvm-1"] +def test_logs_no_stop_for_a_resume_hook_after_the_call_returned( + start: Any, logger: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + # The call has returned, and the worker still closes its client. A + # resume hook now has no attempt left to stop. So it logs no stop. + client = _ClosingClient() + monkeypatch.setattr(worker_module, "create_microvms_client", lambda r: client) + h = start(create_microvms_client=None) + + h.run_hook(IDLE) + h.timers.advance(IDLE) + assert client.closing.wait(WAIT) + assert h.hook("resume") == 200 + client.release_close.set() + h.settle() + + assert client.suspended == ["mvm-1"] + assert stop_lines(logger) == [] + # The resume hook has ended the refusal. So the end of the call starts no + # grace period. + assert h.listener._cancel_grace_timer is None # noqa: SLF001 + assert h.job("cb-1") == 202 + + def test_keeps_a_hook_refusal_after_a_refusal_for_now_of_the_own_call( start: Any, logger: Any ) -> None: @@ -2155,6 +2463,123 @@ def test_refuses_jobs_with_503_after_the_terminate_hook(start: Any) -> None: assert not h.started("cb-1") +def test_answers_a_resent_known_job_with_202_after_the_terminate_hook( + start: Any, logger: Any +) -> None: + # A resend starts nothing. The terminate hook fails the running job's + # callback, and the finished job has reported its outcome. So the durable + # function learns each outcome from the callback. A 503 would only make + # it deliver the job again to a MicroVM that is ending. + h = start() + h.run_hook() + assert h.job("cb-done") == 202 + h.finish_job("cb-done") + assert h.job("cb-running") == 202 + assert eventually(lambda: h.started("cb-running")) + assert h.hook("terminate") == 200 + + assert h.job("cb-running") == 202 + assert h.job("cb-done") == 202 + assert h.job("cb-new") == 503 + h.jobs.release_all() + assert h.worker.idle(WAIT) + + assert h.jobs.started() == ["cb-done", "cb-running"] + assert eventually( + lambda: logger.messages("warning").count("duplicate job ignored") == 2 + ) + assert h.lambda_client.recorded() == [ + {"kind": "success", "callbackId": "cb-done"}, + terminated_failure("cb-running"), + ] + + +@pytest.mark.parametrize("refusal", ["own-suspend", "suspend-hook"]) +def test_answers_a_resent_finished_job_with_202_during_a_suspend_refusal( + start: Any, logger: Any, refusal: str +) -> None: + # The job has ended, so a resend starts nothing and cannot freeze with + # the MicroVM. A 503 would make the durable function resume the MicroVM + # only to get this answer. + h = start() + h.run_hook(IDLE) + assert h.job("cb-done") == 202 + h.finish_job("cb-done") + if refusal == "own-suspend": + h.advance(IDLE) + assert h.microvms.suspended == ["mvm-1"] + else: + assert h.hook("suspend") == 200 + + assert h.job("cb-done") == 202 + assert h.job("cb-new") == 503 + + assert h.jobs.started() == ["cb-done"] + assert eventually( + lambda: logger.messages("warning").count("duplicate job ignored") == 1 + ) + # The duplicate did not end the refusal. + assert h.job("cb-new") == 503 + + +def test_refuses_a_resent_running_job_with_503_during_a_suspend_refusal( + start: Any, logger: Any +) -> None: + # The running job can freeze with the MicroVM. A 202 would end the + # delivery, and nothing would resume the MicroVM. After the 503, the + # durable function checks the state, resumes the MicroVM, and delivers + # the job again. + h = start() + h.run_hook(IDLE) + assert h.job("cb-1") == 202 + assert eventually(lambda: h.started("cb-1")) + assert h.hook("suspend") == 200 + + assert h.job("cb-1") == 503 + assert h.job("cb-2") == 503 + assert h.jobs.started() == ["cb-1"] + assert logger.messages("warning").count("duplicate job ignored") == 0 + assert logger.messages("info").count("job refused: the MicroVM is suspending") == 2 + + assert h.hook("resume") == 200 + assert h.job("cb-1") == 202 + h.finish_job("cb-1") + + assert h.jobs.started() == ["cb-1"] + assert eventually( + lambda: h.lambda_client.recorded() + == [{"kind": "success", "callbackId": "cb-1"}] + ) + assert eventually( + lambda: logger.messages("warning").count("duplicate job ignored") == 1 + ) + + +def test_answers_a_resent_running_job_with_202_after_suspend_and_terminate_hooks( + start: Any, logger: Any +) -> None: + # The terminate hook fails the running job's callback. So the durable + # function learns the outcome from the callback, also when a suspend + # refusal is still in effect. + h = start() + h.run_hook() + assert h.job("cb-1") == 202 + assert eventually(lambda: h.started("cb-1")) + assert h.hook("suspend") == 200 + assert h.hook("terminate") == 200 + + assert h.job("cb-1") == 202 + assert h.job("cb-2") == 503 + h.jobs.release_all() + assert h.worker.idle(WAIT) + + assert h.jobs.started() == ["cb-1"] + assert eventually( + lambda: logger.messages("warning").count("duplicate job ignored") == 1 + ) + assert h.lambda_client.recorded() == [terminated_failure("cb-1")] + + def test_never_suspends_after_the_terminate_hook(start: Any) -> None: h = start() h.run_hook(IDLE) diff --git a/packages/aws-durable-execution-sdk-python-microvm-worker/tests/self_suspend_test.py b/packages/aws-durable-execution-sdk-python-microvm-worker/tests/self_suspend_test.py index 671e89137..71dc12b3b 100644 --- a/packages/aws-durable-execution-sdk-python-microvm-worker/tests/self_suspend_test.py +++ b/packages/aws-durable-execution-sdk-python-microvm-worker/tests/self_suspend_test.py @@ -20,6 +20,7 @@ from botocore.config import Config from botocore.exceptions import ( ClientError, + ConnectionClosedError, EndpointConnectionError, NoCredentialsError, ) @@ -29,12 +30,16 @@ SUSPEND_CALL_TIMEOUT_SECONDS, SuspendDeadlineError, SuspendOutcome, + SuspendResumedError, classify_suspend_error, create_microvms_client, suspend_microvm, ) +_BEFORE_SEND = "before-send.lambda-microvms.SuspendMicrovm" + + def suspend_error(code: str, status: int, retries: int = 0) -> ClientError: response: Any = { "Error": {"Code": code, "Message": f"{code} message"}, @@ -93,6 +98,9 @@ def test_classifies_a_first_attempt_that_can_succeed_later_as_refused_for_now( suspend_error("InternalServerException", 500), TimeoutError("timed out"), SuspendDeadlineError("passed"), + # A resume hook stopped a later attempt. An earlier attempt may have + # suspended the MicroVM. + SuspendResumedError("resumed"), EndpointConnectionError(endpoint_url="https://example.com"), RuntimeError("boom"), ], @@ -169,7 +177,9 @@ def test_the_default_client_ends_a_read_at_the_call_timeout() -> None: assert client.meta.region_name == "us-west-2" assert config.read_timeout == SUSPEND_CALL_TIMEOUT_SECONDS assert config.connect_timeout == 5 - assert config.retries == {"mode": "standard", "total_max_attempts": 3} + # One attempt: a retry after a lost answer can run after the MicroVM + # resumed, and suspend it with the next job inside. + assert config.retries == {"mode": "standard", "total_max_attempts": 1} finally: client.close() @@ -212,36 +222,40 @@ def send(**_: Any) -> AWSResponse: @pytest.mark.parametrize( - ("status", "code", "headers", "attempts", "outcome"), + ("status", "code", "headers", "outcome"), [ - # A throttle is retried. So its error reports 2 retries, and an - # earlier attempt may have reached the service. - (429, "ThrottlingException", {}, 3, SuspendOutcome.UNKNOWN), - # A signature error with a clock offset is not retried. botocore does - # not correct the clock from the server's Date header. + # The service did nothing. The worker calls again after the next + # idle time. + (429, "ThrottlingException", {}, SuspendOutcome.REFUSED_FOR_NOW), + # A server error does not show whether the service acted. + (500, "InternalServerException", {}, SuspendOutcome.UNKNOWN), + (503, "ServiceUnavailableException", {}, SuspendOutcome.UNKNOWN), + # botocore does not correct the clock from the server's Date header. ( 403, "InvalidSignatureException", {"Date": formatdate(time.time() + 600, usegmt=True)}, - 1, SuspendOutcome.REFUSED_FOR_NOW, ), - (403, "InvalidSignatureException", {}, 1, SuspendOutcome.REFUSED_FOR_NOW), - (403, "ExpiredTokenException", {}, 1, SuspendOutcome.REFUSED_FOR_NOW), + (403, "InvalidSignatureException", {}, SuspendOutcome.REFUSED_FOR_NOW), + (403, "ExpiredTokenException", {}, SuspendOutcome.REFUSED_FOR_NOW), + ], + ids=[ + "throttle", + "server-error", + "unavailable", + "signature-clock-offset", + "signature-mismatch", + "expired-token", ], - ids=["throttle", "signature-clock-offset", "signature-mismatch", "expired-token"], ) -def test_the_default_client_retries_a_throttle_and_no_signature_or_token_error( +def test_the_default_client_makes_one_attempt_for_an_error_answer( credentials: None, - monkeypatch: pytest.MonkeyPatch, status: int, code: str, headers: dict[str, str], - attempts: int, outcome: SuspendOutcome, ) -> None: - # No backoff, so that the retries do not slow the test. - monkeypatch.setattr(ExponentialBackoff, "delay_amount", lambda self, context: 0) client = create_microvms_client("us-east-1") try: sent = _answer_every_attempt(client, status, code, **headers) @@ -249,14 +263,127 @@ def test_the_default_client_retries_a_throttle_and_no_signature_or_token_error( with pytest.raises(ClientError) as raised: suspend_microvm(client, "mvm-1", 5.0) - assert len(sent) == attempts + assert len(sent) == 1 metadata = raised.value.response["ResponseMetadata"] - assert metadata["RetryAttempts"] == attempts - 1 + assert metadata["RetryAttempts"] == 0 assert classify_suspend_error(raised.value) is outcome finally: client.close() +def test_the_default_client_makes_one_attempt_after_a_connection_error( + credentials: None, +) -> None: + # botocore retries a closed connection in its standard mode. The default + # client makes no second attempt, and the outcome is unknown. + client = create_microvms_client("us-east-1") + try: + sent: list[int] = [] + + def close(**_: Any) -> None: + sent.append(1) + raise ConnectionClosedError(endpoint_url="https://example.com") + + client.meta.events.register(_BEFORE_SEND, close) + + with pytest.raises(ConnectionClosedError) as raised: + suspend_microvm(client, "mvm-1", 5.0) + + assert sent == [1] + assert classify_suspend_error(raised.value) is SuspendOutcome.UNKNOWN + finally: + client.close() + + +def test_a_retrying_client_retries_a_throttle_and_its_error_is_unknown( + credentials: None, monkeypatch: pytest.MonkeyPatch +) -> None: + # A caller's client can retry. Its error then reports the retries, and an + # earlier attempt may have reached the service. + monkeypatch.setattr(ExponentialBackoff, "delay_amount", lambda self, context: 0) + client = _retrying_client() + try: + sent = _answer_every_attempt(client, 429, "ThrottlingException") + + with pytest.raises(ClientError) as raised: + suspend_microvm(client, "mvm-1", 5.0) + + assert len(sent) == 3 + assert raised.value.response["ResponseMetadata"]["RetryAttempts"] == 2 + assert classify_suspend_error(raised.value) is SuspendOutcome.UNKNOWN + finally: + client.close() + + +def _retrying_client() -> Any: + """A caller's client that makes 3 attempts.""" + return boto3.session.Session().client( + "lambda-microvms", + region_name="us-east-1", + config=Config(retries={"mode": "standard", "total_max_attempts": 3}), + ) + + +def test_starts_no_attempt_after_the_microvm_resumed( + credentials: None, monkeypatch: pytest.MonkeyPatch +) -> None: + # 1. The first attempt reaches the service, and the service suspends the + # MicroVM. + # 2. The freeze cuts the connection, so the answer is lost. + # 3. The resume hook arrives, and the worker accepts jobs again. + # 4. botocore then retries the closed connection. That attempt would + # suspend the MicroVM with a new job inside. + # So the handler stops the retry once resumed() is True. + monkeypatch.setattr(ExponentialBackoff, "delay_amount", lambda self, context: 0) + client = _retrying_client() + resumes = 0 + try: + sent: list[int] = [] + + def land_then_lose_the_answer(**_: Any) -> None: + nonlocal resumes + sent.append(1) + resumes += 1 + raise ConnectionClosedError(endpoint_url="https://example.com") + + client.meta.events.register(_BEFORE_SEND, land_then_lose_the_answer) + + with pytest.raises(SuspendResumedError): + suspend_microvm(client, "mvm-1", 5.0, resumed=lambda: resumes > 0) + + assert sent == [1] + assert _deadline_handlers(client) == [] + finally: + client.close() + + +def test_starts_no_first_attempt_when_the_microvm_resumed_before_the_call() -> None: + events = _RecordingEvents() + client: Any = _EmittingClient(events) + + with pytest.raises(SuspendResumedError): + suspend_microvm(client, "mvm-1", 5.0, resumed=lambda: True) + + assert client.sent == 0 + assert events.handlers == {} + + +def test_starts_the_attempt_while_the_microvm_has_not_resumed() -> None: + events = _RecordingEvents() + client: Any = _EmittingClient(events) + checks: list[int] = [] + + def resumed() -> bool: + checks.append(1) + return False + + suspend_microvm(client, "mvm-1", 5.0, resumed=resumed) + + assert client.sent == 1 + assert checks == [1] + assert events.handlers == {} + + class _SilentService: """A local HTTP service that accepts connections and never answers, and records when each attempt arrives.""" @@ -427,6 +554,21 @@ def unregister(self, event: str, unique_id: str) -> None: self.handlers.pop(unique_id, None) +class _EmittingClient: + """A client that runs the registered handlers before its one attempt, + as botocore emits ``before-send``.""" + + def __init__(self, events: _RecordingEvents) -> None: + self.meta = type("Meta", (), {"events": events})() + self.sent = 0 + + def suspend_microvm(self, **kwargs: Any) -> dict[str, Any]: + for handler in list(self.meta.events.handlers.values()): + handler(request=None) + self.sent += 1 + return {} + + class _HeldClient: """A client whose calls wait until two of them run at once."""