Skip to content

fix(microvm-worker): log timer failures after lock #838

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ParidelPooya merged 2 commits into from
Oct 25, 2026
Merged

fix(microvm-worker): log timer failures after lock #838

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d25e65c
fix(microvm-worker): log timer failures after lock
ParidelPooya Oct 25, 2026
781c8e3
fix(microvm-worker): guard the timer lock rule
ParidelPooya Oct 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 83 additions & 22 deletions ...n-sdk-python-microvm-worker/src/aws_durable_execution_sdk_python_microvm_worker/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -405,6 +406,12 @@ 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 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.
Expand Down Expand Up @@ -690,7 +697,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
Expand All @@ -703,7 +710,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()
Expand Down Expand Up @@ -804,7 +811,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
Expand Down Expand Up @@ -1036,7 +1043,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()
Expand All @@ -1054,7 +1061,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
Expand Down Expand Up @@ -1302,6 +1309,58 @@ 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 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.

Expand All @@ -1312,8 +1371,10 @@ 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._check_locked()
self._stop_idle_time()
auto_suspend = self._auto_suspend
if (
Expand Down Expand Up @@ -1346,9 +1407,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."""
Expand Down Expand Up @@ -1393,7 +1452,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)
Expand All @@ -1404,7 +1463,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(
Expand All @@ -1423,7 +1482,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
Expand Down Expand Up @@ -1489,7 +1548,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
Expand Down Expand Up @@ -1520,7 +1579,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.
Expand Down Expand Up @@ -1572,8 +1631,10 @@ 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._check_locked()
self._cancel_grace()
generation = self._grace_generation
try:
Expand All @@ -1582,9 +1643,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:
Expand All @@ -1595,7 +1655,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
Expand All @@ -1605,8 +1665,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()
Expand Down
Loading
Loading