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 07e3826c9..df64053bd 100644 --- a/packages/aws-durable-execution-sdk-python-microvm-worker/README.md +++ b/packages/aws-durable-execution-sdk-python-microvm-worker/README.md @@ -112,6 +112,7 @@ A session's run hook payload can set `autoSuspendIdleSeconds`. The session sets 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. `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. 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 cb2704b0e..e6d5d0e3b 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 @@ -178,135 +178,26 @@ def _start_timer(delay: float, action: Callable[[], None]) -> Callable[[], None] return timer.cancel -def _failed_call_warning( - outcome: SuspendOutcome, - *, - terminating: bool, - closed: bool, - suspend_hook_arrived: bool, - suspending: bool, -) -> str: - """Return the warning for a failed SuspendMicrovm call. - - The text depends on the outcome and on the listener state when the - outcome is handled. The caller reads that state under the lock, before - it changes it. The conditions, their order, and the texts match the - JavaScript worker's ``failedCallWarning``. - - 1. A terminate hook during the call: ``_start_idle_time`` starts no - idle time. So the listener never calls again. A rejection keeps its - permission hint. For an unknown outcome, the text also says that the - listener refuses jobs, because a terminating listener refuses jobs - for good. - 2. A rejection with a suspend hook during the call, while the hook's - refusal is still in effect: the MicroVM is suspending anyway. So the - refusal stays until the resume hook, or 30 seconds after the last - suspend hook. The text keeps the permission hint. - 3. A rejection otherwise: the MicroVM keeps running. - 4. A refusal for now after ``close()`` during the call: - ``_start_idle_time`` starts no idle time. So the listener never - calls again. - 5. A refusal for now with a suspend hook during the call, while the - hook's refusal is still in effect: the refusal stays, as in item 2. - 6. A refusal for now with a suspend hook during the call, after the - hook's refusal ended: the listener accepts jobs, and calls again one - idle time after the call and every job end. - 7. A refusal for now otherwise: the next idle time lasts at least 30 - seconds, unless a job or a resume hook starts the idle time again. - 8. An unknown outcome after the refusal of jobs ended during the call: - a resume hook ended it, or the 30-second limit of a suspend hook's - refusal. The listener starts no new refusal after an unknown outcome - when ``suspending`` is false. So it accepts jobs. It calls again one - idle time after the call and every job end. - 9. The same after ``close()`` during the call: the listener accepts - jobs. ``_start_idle_time`` starts no idle time. So the listener never - calls again. - 10. An unknown outcome otherwise: the refusal continues until the resume - hook, or for 30 seconds. - """ - if outcome is SuspendOutcome.UNKNOWN: - if terminating: - return ( - "the suspend call for the idle MicroVM had an unknown outcome. " - "The MicroVM is terminating, so the worker refuses jobs and " - "does not try the call again." - ) - if not suspending: - if closed: - return ( - "the suspend call for the idle MicroVM had an unknown " - "outcome. The refusal of jobs already ended during the " - "call, on a resume hook or after 30 seconds. So the worker " - "accepts jobs. The worker is closed, so it does not try " - "the call again." - ) - return ( - "the suspend call for the idle MicroVM had an unknown outcome. " - "The refusal of jobs already ended during the call, on a " - "resume hook or after 30 seconds. So the worker accepts jobs. " - "It tries the call again one idle time after the call and " - "every job end." - ) - return ( - "the suspend call for the idle MicroVM had an unknown outcome. The " - "worker refuses jobs until the MicroVM resumes, or for 30 seconds." - ) - if outcome is SuspendOutcome.REJECTED: - if terminating: - return ( - "could not suspend the idle MicroVM, because the service " - "rejected the call. The MicroVM is terminating, so the worker " - "does not try the call again. A missing permission needs " - "lambda:SuspendMicrovm on the MicroVM image in its execution " - "role." - ) - if suspend_hook_arrived and suspending: - return ( - "could not suspend the idle MicroVM, because the service " - "rejected the call. A suspend hook arrived during the call, so " - "the MicroVM is suspending anyway. The hook's refusal of jobs " - "stays until the resume hook, or 30 seconds after the last " - "suspend hook. A missing permission needs lambda:SuspendMicrovm " - "on the MicroVM image in its execution role." - ) - return ( - "could not suspend the idle MicroVM, because the service rejected " - "the call. It keeps running. A missing permission needs " - "lambda:SuspendMicrovm on the MicroVM image in its execution role." - ) - if terminating: - return ( - "could not suspend the idle MicroVM, because the service refused " - "the call for now. The MicroVM is terminating, so the worker does " - "not try the call again." - ) - if closed: - return ( - "could not suspend the idle MicroVM, because the service refused " - "the call for now. The worker is closed, so it does not try the " - "call again." - ) - if suspend_hook_arrived and suspending: - return ( - "could not suspend the idle MicroVM, because the service refused " - "the call for now. A suspend hook arrived during the call, so this " - "call does not end the hook's refusal. The idle time starts again " - "after the resume hook, or 30 seconds after the last suspend hook." - ) - if suspend_hook_arrived: - return ( - "could not suspend the idle MicroVM, because the service refused " - "the call for now. A suspend hook arrived during the call, and the " - "hook's refusal of jobs ended before the call ended. So the worker " - "accepts jobs. It tries the call again one idle time after the " - "call and every job end." - ) - return ( +# The warning after a failed SuspendMicrovm call. Each outcome has one name +# and one message. The listener state goes into the warning's fields. The +# names and the messages match the JavaScript worker word for word. +_FAILED_CALL_WARNINGS: Mapping[SuspendOutcome, tuple[str, str]] = { + SuspendOutcome.REJECTED: ( + "rejected", + "could not suspend the idle MicroVM, because the service rejected the " + "call. A missing permission needs lambda:SuspendMicrovm on the MicroVM " + "image in its execution role.", + ), + SuspendOutcome.REFUSED_FOR_NOW: ( + "refusedForNow", "could not suspend the idle MicroVM, because the service refused the " - "call for now. It keeps running. The worker tries the call again after " - "the next idle time. That idle time lasts at least 30 seconds, unless a " - "job or a resume hook starts the idle time again." - ) + "call for now.", + ), + SuspendOutcome.UNKNOWN: ( + "unknown", + "the suspend call for the idle MicroVM had an unknown outcome.", + ), +} @dataclass(frozen=True, kw_only=True) @@ -460,6 +351,18 @@ class MicrovmWorkerListener: 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 listener 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 listener acted on the outcome. + ``closed`` states whether ``close()`` came before the listener acted + on the outcome. ``refusingJobs`` states whether the listener answers + jobs with HTTP 503 after it has acted on the outcome. When the image enables the ``suspend`` and ``terminate`` hooks, the listener acts on them: @@ -1613,52 +1516,50 @@ def _call_suspend( "waitedSeconds": waited_seconds, }, ) + # The block after the client is closed decides whether the refusal + # continues after an unknown outcome. The warning reads its fields in + # that block. So for an unknown outcome, the warning comes after + # close(). + unknown_described: object | None = None try: suspend_microvm(client, target, SUSPEND_CALL_TIMEOUT_SECONDS) except Exception as error: # noqa: BLE001 outcome = classify_suspend_error(error) - deferred = [] - with self._lock: - # The warning depends on this state. The block reads it - # before it changes it. For a rejection and a refusal for - # now, this block also makes the decision. So for these two - # outcomes, the text matches the decision, whatever another - # thread does after the lock is released. For an unknown - # outcome, the block at the end of this method decides - # whether the 30-second refusal starts. See issue #820. - hook_arrived = self._suspend_hooks != hooks_before - terminating = self._terminating - closed = self._closed - suspending = self._suspending - # The service did nothing. So jobs can run again at once. A - # suspend hook that arrived since the idle time ended means - # that the MicroVM is suspending anyway, so its refusal stays. - if outcome is not SuspendOutcome.UNKNOWN and not hook_arrived: - deferred = self._stop_refusing() - # A rejection, such as a missing permission, is not - # retried until a job ends. A refusal for now is retried - # after the next idle time. That idle time lasts at least - # 30 seconds, because the same error can repeat on every - # call. - if outcome is SuspendOutcome.REFUSED_FOR_NOW: - self._start_idle_time(after_refusal=True) - self._logger.warning( - _failed_call_warning( - outcome, - terminating=terminating, - closed=closed, - suspend_hook_arrived=hook_arrived, - suspending=suspending, - ), - extra={**fields, "error": describe(error)}, - ) - if outcome is not SuspendOutcome.UNKNOWN: + # describe() calls str(error). A client from + # create_microvms_client can raise any exception type. So the + # caller of the listener controls __str__(). That method can be + # slow, or can call the listener. A listener method takes the + # lock, and the lock is not reentrant. So the error is described + # here, before either locked block. + described = describe(error) + if outcome is SuspendOutcome.UNKNOWN: + # 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. + unknown_described = described + else: + deferred = [] + with self._lock: + # The service did nothing. So jobs can run again at once. + # A suspend hook that arrived since the idle time ended + # means that the MicroVM is suspending anyway, so its + # refusal stays. + if self._suspend_hooks == hooks_before: + deferred = self._stop_refusing() + # A rejection, such as a missing permission, is not + # retried until a job ends. A refusal for now is + # retried after the next idle time. That idle time + # lasts at least 30 seconds, because the same error + # can repeat on every call. + if outcome is SuspendOutcome.REFUSED_FOR_NOW: + self._start_idle_time(after_refusal=True) + message, extra = self._failed_call_log( + outcome, described, target, hooks_before + ) + self._logger.warning(message, extra=extra) self._launch_deferred(deferred) return - # 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. finally: if owns_client: # Each suspend creates its own client. Closing it releases @@ -1676,6 +1577,47 @@ def _call_suspend( # closed listener does not refuse jobs forever. if self._suspending: self._start_grace() + if unknown_described is None: + return + message, extra = self._failed_call_log( + SuspendOutcome.UNKNOWN, unknown_described, target, hooks_before + ) + self._logger.warning(message, extra=extra) + + def _failed_call_log( + self, + outcome: SuspendOutcome, + described: object, + microvm_id: str, + hooks_before: int, + ) -> tuple[str, dict[str, object]]: + """Return the warning for a failed SuspendMicrovm call, and its fields. + + The message names only the outcome. The fields state whether a + suspend hook arrived since the idle time ended, whether the listener + is terminating or closed, and whether it refuses jobs. + + The caller holds the lock, and has already acted on the outcome in + the same locked block. So the fields state what the listener does + after the outcome. The caller logs the warning after it releases the + lock. + + This method runs under the lock. So it must not call code that the + caller of the listener controls. ``described`` is the result of + :func:`describe`, which calls ``str()`` on the error. So the caller + describes the error before it takes the lock. + """ + name, message = _FAILED_CALL_WARNINGS[outcome] + return message, { + "microvmId": microvm_id, + "error": described, + "outcome": name, + "suspendHookArrived": self._suspend_hooks != hooks_before, + "terminating": self._terminating, + "closed": self._closed, + # The job path refuses a job with HTTP 503 in both states. + "refusingJobs": self._terminating or self._suspending, + } def _start_grace(self) -> None: """End the refusal after the grace period, unless the ``resume`` hook 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 f25f8d6db..2ce2e169a 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 @@ -39,74 +39,81 @@ GRACE = 30.0 TERMINATED_MESSAGE = "the MicroVM was terminated before the job finished" -# The warnings for a failed SuspendMicrovm call. They match the JavaScript -# worker's texts word for word. +# The warning for a failed SuspendMicrovm call has one message per outcome. +# The messages, the outcome names, and the field names match the JavaScript +# worker's word for word. REJECTED = ( "could not suspend the idle MicroVM, because the service rejected the " - "call. It keeps running. A missing permission needs lambda:SuspendMicrovm " - "on the MicroVM image in its execution role." -) -REJECTED_TERMINATING = ( - "could not suspend the idle MicroVM, because the service rejected the " - "call. The MicroVM is terminating, so the worker does not try the call " - "again. A missing permission needs lambda:SuspendMicrovm on the MicroVM " + "call. A missing permission needs lambda:SuspendMicrovm on the MicroVM " "image in its execution role." ) -REJECTED_SUSPENDING = ( - "could not suspend the idle MicroVM, because the service rejected the " - "call. A suspend hook arrived during the call, so the MicroVM is " - "suspending anyway. The hook's refusal of jobs stays until the resume " - "hook, or 30 seconds after the last suspend hook. A missing permission " - "needs lambda:SuspendMicrovm on the MicroVM image in its execution role." -) REFUSED = ( - "could not suspend the idle MicroVM, because the service refused the call " - "for now. It keeps running. The worker tries the call again after the " - "next idle time. That idle time lasts at least 30 seconds, unless a job or " - "a resume hook starts the idle time again." -) -REFUSED_TERMINATING = ( - "could not suspend the idle MicroVM, because the service refused the call " - "for now. The MicroVM is terminating, so the worker does not try the call " - "again." -) -REFUSED_CLOSED = ( - "could not suspend the idle MicroVM, because the service refused the call " - "for now. The worker is closed, so it does not try the call again." -) -REFUSED_SUSPENDING = ( - "could not suspend the idle MicroVM, because the service refused the call " - "for now. A suspend hook arrived during the call, so this call does not " - "end the hook's refusal. The idle time starts again after the resume " - "hook, or 30 seconds after the last suspend hook." -) -REFUSED_AFTER_HOOK = ( - "could not suspend the idle MicroVM, because the service refused the call " - "for now. A suspend hook arrived during the call, and the hook's refusal " - "of jobs ended before the call ended. So the worker accepts jobs. It " - "tries the call again one idle time after the call and every job end." -) -UNKNOWN = ( - "the suspend call for the idle MicroVM had an unknown outcome. The worker " - "refuses jobs until the MicroVM resumes, or for 30 seconds." -) -UNKNOWN_TERMINATING = ( - "the suspend call for the idle MicroVM had an unknown outcome. The " - "MicroVM is terminating, so the worker refuses jobs and does not try the " - "call again." -) -UNKNOWN_ENDED = ( - "the suspend call for the idle MicroVM had an unknown outcome. The refusal " - "of jobs already ended during the call, on a resume hook or after 30 " - "seconds. So the worker accepts jobs. It tries the call again one idle " - "time after the call and every job end." -) -UNKNOWN_ENDED_CLOSED = ( - "the suspend call for the idle MicroVM had an unknown outcome. The refusal " - "of jobs already ended during the call, on a resume hook or after 30 " - "seconds. So the worker accepts jobs. The worker is closed, so it does not " - "try the call again." + "could not suspend the idle MicroVM, because the service refused the call for now." ) +UNKNOWN = "the suspend call for the idle MicroVM had an unknown outcome." +WARNINGS = {"rejected": REJECTED, "refusedForNow": REFUSED, "unknown": UNKNOWN} +WARNING_FIELDS = [ + "microvmId", + "error", + "outcome", + "suspendHookArrived", + "terminating", + "closed", + "refusingJobs", +] + + +def assert_warning( + message: str, + data: dict[str, Any], + outcome: str, + *, + refusing_jobs: bool, + suspend_hook_arrived: bool = False, + terminating: bool = False, + closed: bool = False, +) -> None: + """Assert that one warning for a failed SuspendMicrovm call has this + outcome and these fields.""" + assert message == WARNINGS[outcome] + assert list(data) == WARNING_FIELDS + assert data["microvmId"] == "mvm-1" + assert sorted(data["error"]) == ["message", "name"] + assert {name: data[name] for name in WARNING_FIELDS[2:]} == { + "outcome": outcome, + "suspendHookArrived": suspend_hook_arrived, + "terminating": terminating, + "closed": closed, + "refusingJobs": refusing_jobs, + } + + +def assert_one_warning( + logger: Any, + outcome: str, + *, + refusing_jobs: bool, + suspend_hook_arrived: bool = False, + terminating: bool = False, + closed: bool = False, +) -> dict[str, Any]: + """Assert that the worker logged exactly one warning, for one failed + SuspendMicrovm call, with this outcome and these fields. Returns the + fields.""" + warnings = [(m, data) for level, m, data in logger.lines if level == "warning"] + assert len(warnings) == 1 + message, data = warnings[0] + assert_warning( + message, + data, + outcome, + refusing_jobs=refusing_jobs, + suspend_hook_arrived=suspend_hook_arrived, + terminating=terminating, + closed=closed, + ) + return data + LAMBDA_HEADERS = {"Host": "localhost:8080"} FORWARDED_HEADERS = { @@ -901,11 +908,8 @@ def test_logs_a_rejected_suspend_accepts_jobs_at_once_and_does_not_retry( h.advance(IDLE * 6) assert h.microvms.suspended == ["mvm-1"] - warnings = [line for line in logger.lines if line[0] == "warning"] - assert len(warnings) == 1 - assert warnings[0][1] == REJECTED - assert warnings[0][2]["microvmId"] == "mvm-1" - assert warnings[0][2]["error"]["name"] == "ClientError" + data = assert_one_warning(logger, "rejected", refusing_jobs=False) + assert data["error"]["name"] == "ClientError" assert h.job("cb-1") == 202 @@ -957,7 +961,7 @@ def test_keeps_refusing_jobs_after_an_unknown_outcome( h.run_hook(IDLE) h.advance(IDLE) - assert logger.messages("warning") == [UNKNOWN] + assert_one_warning(logger, "unknown", refusing_jobs=True) assert h.job("cb-1") == 503 h.advance(GRACE) @@ -999,7 +1003,7 @@ def test_accepts_jobs_at_once_and_retries_after_a_refusal_for_now( h.run_hook(IDLE) h.advance(IDLE) - assert logger.messages("warning") == [REFUSED] + assert_one_warning(logger, "refusedForNow", refusing_jobs=False) assert h.job("cb-1") == 202 h.finish_job("cb-1") @@ -1036,9 +1040,12 @@ def test_retries_a_refusal_for_now_after_the_idle_time_or_30_seconds_if_longer( assert h.microvms.suspended == ["mvm-1", "mvm-1"] h.advance(0.001) assert h.microvms.suspended == ["mvm-1", "mvm-1", "mvm-1"] - # One warning per refusal. - refusals = [m for m in logger.messages("warning") if "for now" in m] - assert len(refusals) == 3 + # One warning per call. The service refused each call for now. So after + # each call, the worker accepts jobs again. + warnings = [(m, data) for level, m, data in logger.lines if level == "warning"] + assert len(warnings) == 3 + for message, data in warnings: + assert_warning(message, data, "refusedForNow", refusing_jobs=False) def test_makes_one_call_per_30_seconds_while_a_refusal_for_now_persists( @@ -1227,8 +1234,9 @@ def test_keeps_a_hook_refusal_after_a_refusal_for_now_of_the_own_call( h.settle() assert h.job("cb-1") == 503 - # No 30-second floor applies, so the warning does not promise one. - assert logger.messages("warning") == [REFUSED_SUSPENDING] + assert_one_warning( + logger, "refusedForNow", suspend_hook_arrived=True, refusing_jobs=True + ) def hold_failed_call(h: Harness, failure: BaseException) -> None: @@ -1256,22 +1264,18 @@ def end_failed_call(h: Harness) -> None: ids=["terminate", "suspend-then-terminate", "resume-then-terminate"], ) @pytest.mark.parametrize( - ("failure", "text"), + ("failure", "outcome"), [ - (suspend_error("AccessDeniedException", 403), REJECTED_TERMINATING), - (suspend_error("InvalidSignatureException", 403), REFUSED_TERMINATING), - ( - suspend_error("InternalServerException", 500, retries=2), - UNKNOWN_TERMINATING, - ), + (suspend_error("AccessDeniedException", 403), "rejected"), + (suspend_error("InvalidSignatureException", 403), "refusedForNow"), + (suspend_error("InternalServerException", 500, retries=2), "unknown"), ], ids=["rejected", "refused-for-now", "unknown"], ) def test_never_calls_again_after_a_terminate_hook_during_a_failed_call( - start: Any, logger: Any, failure: BaseException, text: str, hooks: list[str] + start: Any, logger: Any, failure: BaseException, outcome: str, hooks: list[str] ) -> None: # A terminating listener refuses jobs for good, and starts no idle time. - # So the text names the terminate hook, also after a suspend hook. h = start() hold_failed_call(h, failure) for hook in hooks: @@ -1283,20 +1287,25 @@ def test_never_calls_again_after_a_terminate_hook_during_a_failed_call( h.advance(GRACE * 4) assert h.job("cb-2") == 503 assert h.microvms.suspended == ["mvm-1"] - assert logger.messages("warning") == [text] + assert_one_warning( + logger, + outcome, + suspend_hook_arrived="suspend" in hooks, + terminating=True, + refusing_jobs=True, + ) @pytest.mark.parametrize( - ("failure", "text"), + ("failure", "outcome"), [ - (suspend_error("InvalidSignatureException", 403), REFUSED_CLOSED), - # A rejection promises no retry. So its text stays true after close(). - (suspend_error("AccessDeniedException", 403), REJECTED), + (suspend_error("InvalidSignatureException", 403), "refusedForNow"), + (suspend_error("AccessDeniedException", 403), "rejected"), ], ids=["refused-for-now", "rejected"], ) def test_never_calls_again_after_close_during_a_failed_call( - start: Any, logger: Any, failure: BaseException, text: str + start: Any, logger: Any, failure: BaseException, outcome: str ) -> None: h = start() hold_failed_call(h, failure) @@ -1308,10 +1317,10 @@ def test_never_calls_again_after_close_during_a_failed_call( assert h.hook("resume") == 200 h.advance(GRACE * 4) assert h.microvms.suspended == ["mvm-1"] - assert logger.messages("warning") == [text] + assert_one_warning(logger, outcome, closed=True, refusing_jobs=False) -def test_names_the_terminate_hook_when_close_also_came_during_a_refused_call( +def test_never_calls_again_after_close_and_a_terminate_hook_during_a_refused_call( start: Any, logger: Any ) -> None: h = start() @@ -1323,20 +1332,22 @@ def test_names_the_terminate_hook_when_close_also_came_during_a_refused_call( assert h.job("cb-1") == 503 h.advance(GRACE * 4) assert h.microvms.suspended == ["mvm-1"] - assert logger.messages("warning") == [REFUSED_TERMINATING] + assert_one_warning( + logger, "refusedForNow", terminating=True, closed=True, refusing_jobs=True + ) @pytest.mark.parametrize("end", ["resume", "grace"]) @pytest.mark.parametrize( - ("failure", "text"), + ("failure", "outcome"), [ - (suspend_error("AccessDeniedException", 403), REJECTED_SUSPENDING), - (suspend_error("InvalidSignatureException", 403), REFUSED_SUSPENDING), + (suspend_error("AccessDeniedException", 403), "rejected"), + (suspend_error("InvalidSignatureException", 403), "refusedForNow"), ], ids=["rejected", "refused-for-now"], ) def test_keeps_a_suspend_hook_refusal_in_effect_after_a_failed_call( - start: Any, logger: Any, failure: BaseException, text: str, end: str + start: Any, logger: Any, failure: BaseException, outcome: str, end: str ) -> None: # Two suspend hooks arrive during the call, 10 seconds apart. The failed # call does not end their refusal. The refusal ends on the resume hook, @@ -1360,7 +1371,7 @@ def test_keeps_a_suspend_hook_refusal_in_effect_after_a_failed_call( assert h.microvms.suspended == ["mvm-1"] h.advance(0.001) assert h.microvms.suspended == ["mvm-1", "mvm-1"] - assert logger.messages("warning") == [text] + assert_one_warning(logger, outcome, suspend_hook_arrived=True, refusing_jobs=True) @pytest.mark.parametrize("then", ["no-job", "job"]) @@ -1390,11 +1401,13 @@ def test_accepts_jobs_after_a_refusal_for_now_when_the_hook_refusal_ended( assert h.microvms.suspended == ["mvm-1"] h.advance(0.001) assert h.microvms.suspended == ["mvm-1", "mvm-1"] - assert logger.messages("warning") == [REFUSED_AFTER_HOOK] + assert_one_warning( + logger, "refusedForNow", suspend_hook_arrived=True, refusing_jobs=False + ) @pytest.mark.parametrize("end", ["resume", "grace"]) -def test_says_the_microvm_keeps_running_after_a_rejection_when_the_hook_refusal_ended( +def test_accepts_jobs_after_a_rejection_when_the_hook_refusal_ended( start: Any, logger: Any, end: str ) -> None: h = start() @@ -1412,7 +1425,9 @@ def test_says_the_microvm_keeps_running_after_a_rejection_when_the_hook_refusal_ assert h.microvms.suspended == ["mvm-1"] h.advance(0.001) assert h.microvms.suspended == ["mvm-1", "mvm-1"] - assert logger.messages("warning") == [REJECTED] + assert_one_warning( + logger, "rejected", suspend_hook_arrived=True, refusing_jobs=False + ) @pytest.mark.parametrize("then", ["job", "no-job"]) @@ -1430,7 +1445,7 @@ def test_says_the_microvm_keeps_running_after_a_rejection_when_the_hook_refusal_ ], ids=["resume", "suspend-resume", "suspend-grace"], ) -def test_says_the_refusal_ended_after_an_unknown_outcome_when_it_ended_during_the_call( +def test_accepts_jobs_after_an_unknown_outcome_when_the_refusal_ended_during_the_call( start: Any, logger: Any, during: str, failure: BaseException, then: str ) -> None: # The refusal of jobs ends during the call: on a resume hook, or 30 @@ -1454,7 +1469,11 @@ def test_says_the_refusal_ended_after_an_unknown_outcome_when_it_ended_during_th assert logger.messages("warning") == [] end_failed_call(h) - assert logger.messages("warning") == [UNKNOWN_ENDED] + expected = { + "suspend_hook_arrived": during != "resume", + "refusing_jobs": False, + } + assert_one_warning(logger, "unknown", **expected) if then == "job": assert h.job("cb-1") == 202 assert eventually(lambda: h.started("cb-1")) @@ -1465,7 +1484,7 @@ def test_says_the_refusal_ended_after_an_unknown_outcome_when_it_ended_during_th assert h.microvms.suspended == ["mvm-1"] h.advance(0.001) assert h.microvms.suspended == ["mvm-1", "mvm-1"] - assert logger.messages("warning") == [UNKNOWN_ENDED] + assert_one_warning(logger, "unknown", **expected) def test_calls_again_one_idle_time_after_the_last_job_after_an_unknown_outcome( @@ -1481,7 +1500,7 @@ def test_calls_again_one_idle_time_after_the_last_job_after_an_unknown_outcome( assert eventually(lambda: h.started("cb-1")) end_failed_call(h) - assert logger.messages("warning") == [UNKNOWN_ENDED] + assert_one_warning(logger, "unknown", refusing_jobs=False) assert h.job("cb-2") == 202 h.advance(IDLE * 3) assert h.microvms.suspended == ["mvm-1"] @@ -1493,10 +1512,10 @@ def test_calls_again_one_idle_time_after_the_last_job_after_an_unknown_outcome( assert h.microvms.suspended == ["mvm-1"] h.advance(0.001) assert h.microvms.suspended == ["mvm-1", "mvm-1"] - assert logger.messages("warning") == [UNKNOWN_ENDED] + assert_one_warning(logger, "unknown", refusing_jobs=False) -def test_says_that_no_call_follows_an_unknown_outcome_after_close_and_a_resume_hook( +def test_makes_no_call_after_an_unknown_outcome_after_close_and_a_resume_hook( start: Any, logger: Any ) -> None: # A closed listener starts no idle time. So the resume hook ends the @@ -1507,7 +1526,7 @@ def test_says_that_no_call_follows_an_unknown_outcome_after_close_and_a_resume_h assert h.hook("resume") == 200 end_failed_call(h) - assert logger.messages("warning") == [UNKNOWN_ENDED_CLOSED] + assert_one_warning(logger, "unknown", closed=True, refusing_jobs=False) assert h.job("cb-1") == 202 h.finish_job("cb-1") h.advance(IDLE + GRACE * 4) @@ -1515,7 +1534,7 @@ def test_says_that_no_call_follows_an_unknown_outcome_after_close_and_a_resume_h @pytest.mark.parametrize("during", ["no-hook", "suspend", "close"]) -def test_keeps_the_refusal_text_after_an_unknown_outcome_while_the_refusal_lasts( +def test_keeps_refusing_jobs_after_an_unknown_outcome_while_the_refusal_lasts( start: Any, logger: Any, during: str ) -> None: # The refusal is still in effect when the call ends. So the worker @@ -1528,7 +1547,13 @@ def test_keeps_the_refusal_text_after_an_unknown_outcome_while_the_refusal_lasts h.listener.close() end_failed_call(h) - assert logger.messages("warning") == [UNKNOWN] + assert_one_warning( + logger, + "unknown", + suspend_hook_arrived=during == "suspend", + closed=during == "close", + refusing_jobs=True, + ) assert h.job("cb-1") == 503 h.advance(GRACE - 0.001) assert h.job("cb-2") == 503 @@ -1540,6 +1565,137 @@ def test_keeps_the_refusal_text_after_an_unknown_outcome_while_the_refusal_lasts assert h.microvms.suspended == expected +def test_reads_the_unknown_outcome_fields_in_the_block_that_decides_the_refusal( + start: Any, logger: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + # Issue #820. A suspend hook arrives during the call. The call ends with + # an unknown outcome. The hook's 30 seconds then end while the worker + # closes its client. The worker formerly read the refusal state before + # close(), and decided the refusal after close(). So its warning said + # that it refuses jobs, and a job then got HTTP 202. The worker now reads + # the fields in the block after close(), which decides the refusal. + client = _ClosingClient() + client.hold = True + client.failure = ConnectionResetError("reset") + 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 eventually(lambda: client.suspended == ["mvm-1"]) + assert h.hook("suspend") == 200 + h.timers.advance(GRACE - 0.001) + + client.hold = False + client.finish() + assert client.closing.wait(WAIT) + # The hook's 30 seconds end during close(). + h.timers.advance(0.001) + assert logger.messages("warning") == [] + assert h.job("cb-1") == 202 + client.release_close.set() + h.settle() + + assert_one_warning( + logger, "unknown", suspend_hook_arrived=True, refusing_jobs=False + ) + assert h.job("cb-2") == 202 + h.finish_job("cb-1") + h.finish_job("cb-2") + + +class _CallbackError(ClientError): + """A copy of a ClientError whose ``__str__()`` first calls ``callback``. + + A client from create_microvms_client can raise any exception type. So + the caller of the listener controls ``__str__()``. + """ + + def __init__(self, base: ClientError, callback: Callable[[], None]) -> None: + super().__init__(base.response, base.operation_name) + self._callback = callback + + def __str__(self) -> str: + self._callback() + return super().__str__() + + +# One failure for each outcome. A rejection and a refusal for now take the +# lock in the except block. An unknown outcome takes it after close(). +_EACH_OUTCOME = pytest.mark.parametrize( + ("failure", "outcome"), + [ + (suspend_error("AccessDeniedException", 403), "rejected"), + (suspend_error("InvalidSignatureException", 403), "refusedForNow"), + (suspend_error("InternalServerException", 500, retries=2), "unknown"), + ], + ids=["rejected", "refused-for-now", "unknown"], +) + + +@_EACH_OUTCOME +def test_describes_the_call_error_before_it_takes_the_lock( + start: Any, logger: Any, failure: ClientError, outcome: str +) -> None: + # The listener's lock is not reentrant. The error's __str__() calls + # close(), which takes the lock. If the listener described the error + # under the lock, close() would wait for that lock forever. Every hook + # and job request would then wait too. + h = start() + closed = threading.Event() + + def close() -> None: + h.listener.close() + closed.set() + + h.microvms.failure = _CallbackError(failure, close) + h.run_hook(IDLE) + h.timers.advance(IDLE) + if not closed.wait(WAIT): + # threading.Lock lets any thread release it. So the test releases + # the lock that the suspend thread holds. close() then returns, and + # the fixture can close the worker. + h.listener._lock.release() # noqa: SLF001 + closed.wait(WAIT) + pytest.fail("close() in the error's __str__() waited for the lock") + h.settle() + + # Only after an unknown outcome does the refusal continue. + assert_one_warning(logger, outcome, closed=True, refusing_jobs=outcome == "unknown") + + +@_EACH_OUTCOME +def test_answers_a_hook_that_the_call_error_sends( + start: Any, logger: Any, failure: ClientError, outcome: str +) -> None: + # The error's __str__() sends a suspend hook. A request thread handles + # the hook, and takes the lock. If the listener described the error + # under the lock, the hook would wait until its HTTP timeout. + h = start() + statuses: list[int | BaseException] = [] + answered = threading.Event() + + def send_hook() -> None: + try: + statuses.append(h.hook("suspend")) + except Exception as error: # noqa: BLE001 + statuses.append(error) + finally: + answered.set() + + h.microvms.failure = _CallbackError(failure, send_hook) + h.run_hook(IDLE) + h.timers.advance(IDLE) + # The hook's HTTP timeout is WAIT. So this wait ends in either case. + assert answered.wait(2 * WAIT) + assert statuses == [200] + h.settle() + + # The hook arrived before the listener acted on the outcome. So its + # refusal stays. + assert_one_warning(logger, outcome, suspend_hook_arrived=True, refusing_jobs=True) + assert h.job("cb-1") == 503 + + @pytest.mark.parametrize( ("idle", "waited"), [(1.0, GRACE), (45.0, 45.0)], ids=["idle-1s", "idle-45s"] )