From d25e65c991ef7e41f55c1fe19680da9ff14784fd Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Sun, 25 Oct 2026 12:19:39 -0700 Subject: [PATCH 1/2] fix(microvm-worker): log timer failures after lock A timer that could not start was logged under the listener lock. A logger that calls the listener then deadlocked it. Now the failure is recorded under the lock and logged after the lock is released. Fixes #832. --- .../worker.py | 71 +++++++---- .../tests/lifecycle_hooks_test.py | 120 ++++++++++++++++++ 2 files changed, 169 insertions(+), 22 deletions(-) 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 316d2e89f..0d3a699ca 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 @@ -20,7 +20,8 @@ import sys import threading from collections import OrderedDict -from collections.abc import Callable, Mapping +from collections.abc import Callable, Iterator, Mapping +from contextlib import contextmanager from dataclasses import dataclass from functools import cached_property from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -405,6 +406,9 @@ def __init__( # job, a report, a run hook, or a suspend call ends. self._lock = threading.Lock() self._changed = threading.Condition(self._lock) + # The timers that could not start, as a log message and the error. + # _locked() logs them after it releases the lock. + self._timer_failures: list[tuple[str, BaseException]] = [] # The callback IDs of the jobs that run now. self._jobs: set[str] = set() # The callback IDs of jobs that ended, oldest first. @@ -690,7 +694,7 @@ def _handle_suspend_hook(self) -> None: Lambda freezes the MicroVM after the answer. A job accepted now would freeze with it. """ - with self._lock: + with self._locked(): self._suspend_hooks += 1 self._stop_idle_time() self._suspending = True @@ -703,7 +707,7 @@ def _handle_resume_hook(self) -> list[tuple[MicrovmJobDocument, str]]: until the next job ends. Returns the deferred run hook jobs, registered. The caller starts them after the answer. """ - with self._lock: + with self._locked(): if self._suspending: return self._end_suspending() self._start_idle_time() @@ -804,7 +808,7 @@ def _handle_run_hook(self, exchange: _Exchange) -> None: # otherwise both start the job. The body is read before these checks. # So a suspend or terminate hook can arrive while it is still being # read. - with self._lock: + with self._locked(): later = self._run_hook_accepted if not later: self._run_hook_accepted = True @@ -1036,7 +1040,7 @@ def _start_job( # durable function delivers the job again, and the retry may run # it. The run hook fails the callback instead, because Lambda # sends it only once. - with self._changed: + with self._locked(): self._jobs.discard(job.callback_id) self._active.pop(job.callback_id, None) self._start_idle_time() @@ -1054,7 +1058,7 @@ def _run_job( extra={"callbackId": job.callback_id, "error": describe(error)}, ) finally: - with self._changed: + with self._locked(): self._jobs.discard(job.callback_id) self._active.pop(job.callback_id, None) self._finished[job.callback_id] = None @@ -1302,6 +1306,29 @@ def _reporter_for(self, callback_id: str, region: str) -> CallbackReporter: callback_id, self._create_client(region), logger=self._logger ) + @contextmanager + def _locked(self) -> Iterator[None]: + """Hold the lock, then log the timers that could not start. + + Every block that can start a timer takes the lock with this method. + The caller of the listener passes the logger. That logger can be + slow, or can call the listener. A listener method takes the lock, + and the lock is not reentrant. So :meth:`_start_idle_time` and + :meth:`_start_grace` only record a failure under the lock. This + method logs each failure after it releases the lock. It also + describes the error there, as the SuspendMicrovm warning does. It + logs the failures when the block raises too. + """ + self._lock.acquire() + try: + yield + finally: + failures = self._timer_failures + self._timer_failures = [] + self._lock.release() + for message, error in failures: + self._logger.error(message, extra={"error": describe(error)}) + def _start_idle_time(self, *, after_refusal: bool = False) -> None: """Start the idle time, when the session asked for it and no job runs. @@ -1312,7 +1339,8 @@ def _start_idle_time(self, *, after_refusal: bool = False) -> None: call as fast as the client can make it. Every other start uses the session's idle time. - The caller holds the lock. + The caller holds the lock through :meth:`_locked`. A timer that + cannot start is logged after the caller releases the lock. """ self._stop_idle_time() auto_suspend = self._auto_suspend @@ -1346,9 +1374,7 @@ def _start_idle_time(self, *, after_refusal: bool = False) -> None: except Exception as error: # noqa: BLE001 # Without a timer thread, the MicroVM keeps running until a job # ends and the idle time starts again. - self._logger.error( - "could not start the idle time", extra={"error": describe(error)} - ) + self._timer_failures.append(("could not start the idle time", error)) def _stop_idle_time(self) -> None: """Cancel the idle time. The caller holds the lock.""" @@ -1393,7 +1419,7 @@ def _idle_time_ended(self, generation: int, waited_seconds: float) -> None: "could not start the suspend of the idle MicroVM", extra={"microvmId": auto_suspend.microvm_id, "error": describe(error)}, ) - with self._changed: + with self._locked(): deferred = self._end_own_refusal(hooks_before) self._suspend_call_ended() self._launch_deferred(deferred) @@ -1404,7 +1430,7 @@ def _suspend_self( try: self._call_suspend(auto_suspend, hooks_before, waited_seconds) finally: - with self._changed: + with self._locked(): self._suspend_call_ended() def _end_own_refusal( @@ -1423,7 +1449,7 @@ def _end_own_refusal( def _suspend_call_ended(self) -> None: """Count a SuspendMicrovm call as ended, and start the idle time that a resume hook, a job, or a refusal for now asked for during it. The - caller holds the lock.""" + caller holds the lock through :meth:`_locked`.""" self._suspend_calls -= 1 if self._idle_after_call: self._idle_after_call = False @@ -1489,7 +1515,7 @@ def _call_suspend( unknown_described = described else: deferred = [] - with self._lock: + with self._locked(): # 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 @@ -1520,7 +1546,7 @@ def _call_suspend( "could not close the Lambda MicroVMs client", extra={**fields, "error": describe(close_error)}, ) - with self._lock: + with self._locked(): # A resume hook that arrived during the call has already ended # the refusal. The grace timer also runs after close(), so a # closed listener does not refuse jobs forever. @@ -1572,7 +1598,8 @@ def _start_grace(self) -> None: """End the refusal after the grace period, unless the ``resume`` hook ends it first. A timer started earlier is replaced. - The caller holds the lock. + The caller holds the lock through :meth:`_locked`. A timer that + cannot start is logged after the caller releases the lock. """ self._cancel_grace() generation = self._grace_generation @@ -1582,9 +1609,8 @@ def _start_grace(self) -> None: ) except Exception as error: # noqa: BLE001 # Without a timer thread, only the resume hook ends the refusal. - self._logger.error( - "could not start the end of the refusal", - extra={"error": describe(error)}, + self._timer_failures.append( + ("could not start the end of the refusal", error) ) def _cancel_grace(self) -> None: @@ -1595,7 +1621,7 @@ def _cancel_grace(self) -> None: self._cancel_grace_timer = None def _grace_ended(self, generation: int) -> None: - with self._lock: + with self._locked(): if generation != self._grace_generation: return self._cancel_grace_timer = None @@ -1605,8 +1631,9 @@ def _grace_ended(self, generation: int) -> None: def _end_suspending(self) -> list[tuple[MicrovmJobDocument, str]]: """End the refusal of jobs, and start the idle time again. - The caller holds the lock, and the listener is suspending. The caller - starts the returned jobs after it releases the lock. + The caller holds the lock through :meth:`_locked`, and the listener + is suspending. The caller starts the returned jobs after it releases + the lock. """ deferred = self._stop_refusing() self._start_idle_time() 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 2ce2e169a..b7cc42c20 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 @@ -2680,6 +2680,126 @@ def test_refuses_jobs_until_the_resume_hook_when_the_refusal_timer_cannot_start( assert h.job("cb-1") == 202 +class _CallingLogger: + """Records the worker's log lines. The first error line calls ``on_error``. + + Only the first error line calls it. A hook that ``on_error`` sends can + fail to start a timer too, and log a second error line. + """ + + def __init__(self) -> None: + self.lines: list[tuple[str, str, dict[str, Any]]] = [] + self.on_error: Callable[[], None] = lambda: None + self._called = False + self._lock = threading.Lock() + + def _add(self, level: str, message: str, data: Any) -> bool: + with self._lock: + self.lines.append((level, message, dict(data or {}))) + first = level == "error" and not self._called + if first: + self._called = True + return first + + def info(self, msg: object, *args: object, extra: Any = None) -> None: + self._add("info", str(msg), extra) + + def warning(self, msg: object, *args: object, extra: Any = None) -> None: + self._add("warning", str(msg), extra) + + def error(self, msg: object, *args: object, extra: Any = None) -> None: + if self._add("error", str(msg), extra): + self.on_error() + + def errors(self) -> list[tuple[str, dict[str, Any]]]: + with self._lock: + return [(m, data) for level, m, data in self.lines if level == "error"] + + +@dataclass +class _TimerCase: + """A hook that starts one timer, after the run hook ``idle_seconds``.""" + + idle_seconds: float | None + hook: str + message: str + + +# The resume hook starts the idle time. The suspend hook starts the timer +# that ends the refusal. +_EACH_TIMER = pytest.mark.parametrize( + "case", + [ + _TimerCase(IDLE, "resume", "could not start the idle time"), + _TimerCase(None, "suspend", "could not start the end of the refusal"), + ], + ids=["idle-time", "grace"], +) + +_THREAD_ERROR = {"name": "RuntimeError", "message": "can't start new thread"} + + +@_EACH_TIMER +def test_logs_a_timer_failure_after_it_releases_the_lock( + start: Any, timers: FakeTimers, case: _TimerCase +) -> None: + # The listener's lock is not reentrant. The logger's error() calls + # close(), which takes the lock. If the listener logged the failure + # under the lock, close() would wait for that lock forever. Every hook + # and job request would then wait too. + logger = _CallingLogger() + h = start(logger=logger) + assert h.run_hook(case.idle_seconds) == 200 + closed = threading.Event() + + def close() -> None: + h.listener.close() + closed.set() + + logger.on_error = close + timers.failure = RuntimeError("can't start new thread") + hook = h.hook_in_background(case.hook) + if not closed.wait(WAIT): + # threading.Lock lets any thread release it. So the test releases + # the lock that the hook's 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 logger's error() waited for the lock") + assert hook.result(WAIT) == 200 + + assert logger.errors() == [(case.message, {"error": _THREAD_ERROR})] + + +@_EACH_TIMER +def test_answers_a_hook_that_the_timer_failure_log_sends( + start: Any, timers: FakeTimers, case: _TimerCase +) -> None: + # The logger's error() sends a resume hook. A request thread handles the + # hook, and takes the lock. If the listener logged the failure under the + # lock, the resume hook would wait until its HTTP timeout. + logger = _CallingLogger() + h = start(logger=logger) + assert h.run_hook(case.idle_seconds) == 200 + statuses: list[int | BaseException] = [] + + def send_hook() -> None: + try: + statuses.append(h.hook("resume")) + except Exception as error: # noqa: BLE001 + statuses.append(error) + + logger.on_error = send_hook + timers.failure = RuntimeError("can't start new thread") + hook = h.hook_in_background(case.hook) + # The resume hook's HTTP timeout is WAIT. So this wait ends in either + # case. + assert hook.result(2 * WAIT) == 200 + assert statuses == [200] + + assert logger.errors()[0] == (case.message, {"error": _THREAD_ERROR}) + + def test_fails_a_deferred_run_hook_job_whose_thread_cannot_start( start: Any, logger: Any, monkeypatch: pytest.MonkeyPatch ) -> None: From 781c8e38436e581c5f73d092a68ea919bd007ba7 Mon Sep 17 00:00:00 2001 From: Pooya Paridel Date: Sun, 25 Oct 2026 13:22:06 -0700 Subject: [PATCH 2/2] fix(microvm-worker): guard the timer lock rule This addresses the review of #838. A timer start now raises RuntimeError outside _locked(), and the tests and docstrings are fixed. --- .../worker.py | 40 ++++++++++++- .../tests/lifecycle_hooks_test.py | 56 +++++++++++++++---- 2 files changed, 81 insertions(+), 15 deletions(-) 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 0d3a699ca..3e891e580 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 @@ -409,6 +409,9 @@ def __init__( # The timers that could not start, as a log message and the error. # _locked() logs them after it releases the lock. self._timer_failures: list[tuple[str, BaseException]] = [] + # The number of _locked() blocks that run now. It is 0 or 1, because + # the lock is not reentrant. A timer start checks it. + self._locked_depth = 0 # The callback IDs of the jobs that run now. self._jobs: set[str] = set() # The callback IDs of jobs that ended, oldest first. @@ -1315,20 +1318,49 @@ def _locked(self) -> Iterator[None]: slow, or can call the listener. A listener method takes the lock, and the lock is not reentrant. So :meth:`_start_idle_time` and :meth:`_start_grace` only record a failure under the lock. This - method logs each failure after it releases the lock. It also - describes the error there, as the SuspendMicrovm warning does. It - logs the failures when the block raises too. + method logs each failure after it releases the lock. It calls + describe() after the release too, because describe() calls str() on + the error. It logs the failures when the block raises too. + + A block must not call this method again. The lock is not reentrant. + So the second call would wait forever for the lock that the first + call holds. + + This method raises ``_locked_depth`` under the lock, and lowers it + before the release. The two timer starts check it through + :meth:`_check_locked`. """ self._lock.acquire() + self._locked_depth += 1 try: yield finally: + self._locked_depth -= 1 failures = self._timer_failures self._timer_failures = [] self._lock.release() for message, error in failures: self._logger.error(message, extra={"error": describe(error)}) + def _check_locked(self) -> None: + """Raise RuntimeError unless a :meth:`_locked` block runs. + + A timer start records a failure in ``_timer_failures``. Only + :meth:`_locked` logs that list. Suppose a block takes ``self._lock`` + directly and starts a timer that fails. Then the next + :meth:`_locked` block logs the failure, possibly on another thread. + If no such block runs, the failure is never logged. So each timer + start calls this method first. + + Under a plain ``with self._lock:`` block, no other thread can run a + :meth:`_locked` block. So the depth is 0, and this method raises. + The check is an explicit raise, because ``python -O`` removes an + assert. + """ + if self._locked_depth == 0: + msg = "a timer start needs the lock through _locked()" + raise RuntimeError(msg) + def _start_idle_time(self, *, after_refusal: bool = False) -> None: """Start the idle time, when the session asked for it and no job runs. @@ -1342,6 +1374,7 @@ def _start_idle_time(self, *, after_refusal: bool = False) -> None: The caller holds the lock through :meth:`_locked`. A timer that cannot start is logged after the caller releases the lock. """ + self._check_locked() self._stop_idle_time() auto_suspend = self._auto_suspend if ( @@ -1601,6 +1634,7 @@ def _start_grace(self) -> None: The caller holds the lock through :meth:`_locked`. A timer that cannot start is logged after the caller releases the lock. """ + self._check_locked() self._cancel_grace() generation = self._grace_generation try: 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 b7cc42c20..01463abbf 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 @@ -415,9 +415,15 @@ class Harness: def listener(self) -> MicrovmWorkerListener: return self.worker._listener # noqa: SLF001 - def post(self, path: str, body: Any, headers: dict[str, str] | None = None) -> int: + def post( + self, + path: str, + body: Any, + headers: dict[str, str] | None = None, + timeout: float = WAIT, + ) -> int: connection = http.client.HTTPConnection( - host_of(self.worker), self.worker.port, timeout=WAIT + host_of(self.worker), self.worker.port, timeout=timeout ) try: connection.request( @@ -448,17 +454,25 @@ def run_hook( {"microvmId": microvm_id, "runHookPayload": json.dumps(payload)}, ) - def hook(self, name: str, headers: dict[str, str] | None = None) -> int: - return self.post(f"{HOOK_PATH_PREFIX}{name}", {}, headers) + def hook( + self, + name: str, + headers: dict[str, str] | None = None, + timeout: float = WAIT, + ) -> int: + return self.post(f"{HOOK_PATH_PREFIX}{name}", {}, headers, timeout) def hook_in_background( - self, name: str, headers: dict[str, str] | None = None + self, + name: str, + headers: dict[str, str] | None = None, + timeout: float = WAIT, ) -> Future[int]: future: Future[int] = Future() def run() -> None: try: - future.set_result(self.hook(name, headers)) + future.set_result(self.hook(name, headers, timeout)) except BaseException as error: # noqa: BLE001 future.set_exception(error) @@ -486,6 +500,8 @@ def finish_job(self, callback_id: str) -> None: # The job's thread removes the job and starts the idle time in one # locked block. The check above reads without the lock, so it can see # the removal first. Taking the lock waits for the rest of the block. + # The listener writes a timer-failure log line after it releases the + # lock. So that line can come after this method returns. with self.listener._lock: # noqa: SLF001 pass @@ -495,7 +511,8 @@ def settle(self) -> None: # The call's thread lowers the count and starts the idle time in one # locked block. The check above reads without the lock, so it can see # the lower count first. Taking the lock waits for the rest of the - # block. + # block. The listener writes a timer-failure log line after it + # releases the lock. So that line can come after this method returns. with self.listener._lock: # noqa: SLF001 pass @@ -2718,7 +2735,7 @@ def errors(self) -> list[tuple[str, dict[str, Any]]]: @dataclass class _TimerCase: - """A hook that starts one timer, after the run hook ``idle_seconds``.""" + """A hook that starts one timer, after a run hook with ``idle_seconds``.""" idle_seconds: float | None hook: str @@ -2791,15 +2808,30 @@ def send_hook() -> None: logger.on_error = send_hook timers.failure = RuntimeError("can't start new thread") - hook = h.hook_in_background(case.hook) - # The resume hook's HTTP timeout is WAIT. So this wait ends in either - # case. - assert hook.result(2 * WAIT) == 200 + # The inner resume hook's HTTP timeout is WAIT. The outer hook's HTTP + # timeout is 2 * WAIT. Suppose the inner hook waits for the lock. Then + # the inner hook times out first, and the outer hook still gets its + # answer. So the check of statuses reports the failure. Each wait below + # ends within 3 * WAIT. + hook = h.hook_in_background(case.hook, timeout=2 * WAIT) + assert hook.result(3 * WAIT) == 200 assert statuses == [200] assert logger.errors()[0] == (case.message, {"error": _THREAD_ERROR}) +@pytest.mark.parametrize("method", ["_start_idle_time", "_start_grace"]) +def test_a_timer_start_under_the_plain_lock_raises(start: Any, method: str) -> None: + # Only _locked() logs a timer failure. A timer start under a plain + # ``with self._lock:`` block would leave its failure for an unrelated + # block to log. So the timer start refuses to run there. + h = start() + assert h.run_hook(IDLE) == 200 + with h.listener._lock: # noqa: SLF001 + with pytest.raises(RuntimeError, match=r"needs the lock through _locked\(\)"): + getattr(h.listener, method)() + + def test_fails_a_deferred_run_hook_job_whose_thread_cannot_start( start: Any, logger: Any, monkeypatch: pytest.MonkeyPatch ) -> None: