From 622bb2f2132b11cfbda17f455f6b9169a17c26ab Mon Sep 17 00:00:00 2001 From: Vivek Raja Date: Wed, 14 Oct 2026 02:53:43 +0530 Subject: [PATCH] fix(microvm-worker): fail a job whose heartbeats cannot start A job that has its 202, or its 200 from the run hook, gets no second delivery. When the heartbeat thread could not start, the worker only logged the error and recorded the callback ID as finished, so the durable function waited for its heartbeat timeout. The worker now fails the callback, as it does when the job's own thread cannot start. A Content-Length of more than 4,300 digits matched the header pattern, and int() raised past Python's limit on digits. With Expect: 100-continue the error escaped the handler, and the client got no answer. The pattern now allows at most 19 digits, so such a header is invalid and gets HTTP 400. Co-Authored-By: Claude Opus 5.5 --- .../worker.py | 29 +++++++---- .../tests/worker_test.py | 48 +++++++++++++++++++ 2 files changed, 69 insertions(+), 8 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 8ffbe8305..66cdcb35b 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 @@ -101,7 +101,9 @@ _UNKNOWN_MICROVM_ID = "unknown" -_CONTENT_LENGTH = re.compile(r"[0-9]+") +# At most 19 digits, so that int() stays under Python's limit on digits, and +# any declared length still exceeds every body limit. +_CONTENT_LENGTH = re.compile(r"[0-9]{1,19}") _CHUNK_SIZE = re.compile(rb"[0-9A-Fa-f]+") @@ -624,13 +626,24 @@ def on_callback_gone(error: BaseException) -> None: gone.append(error) cancelled.set() - heartbeats = Heartbeats.start( - reporter, - heartbeat_timeout_seconds=job.heartbeat_timeout_seconds, - on_callback_gone=on_callback_gone, - heartbeat_interval_seconds=self._heartbeat_interval_seconds, - logger=self._logger, - ) + try: + heartbeats = Heartbeats.start( + reporter, + heartbeat_timeout_seconds=job.heartbeat_timeout_seconds, + on_callback_gone=on_callback_gone, + heartbeat_interval_seconds=self._heartbeat_interval_seconds, + logger=self._logger, + ) + except Exception as error: # noqa: BLE001 + # The job has its answer, so no later delivery runs it. Fail + # the callback, instead of leaving the job to its heartbeat + # timeout. + self._logger.error( + "could not start the job's heartbeats", + extra={**fields, "error": describe(error)}, + ) + self._report(reporter, _Outcome(error=error), fields) + return try: self._logger.info("job started", extra=fields) context = MicrovmJobContext( diff --git a/packages/aws-durable-execution-sdk-python-microvm-worker/tests/worker_test.py b/packages/aws-durable-execution-sdk-python-microvm-worker/tests/worker_test.py index 18c87e039..a3e24cd0f 100644 --- a/packages/aws-durable-execution-sdk-python-microvm-worker/tests/worker_test.py +++ b/packages/aws-durable-execution-sdk-python-microvm-worker/tests/worker_test.py @@ -580,6 +580,24 @@ def test_answers_400_to_a_body_over_the_limit_without_reading_it(start: Any) -> assert b"request body exceeds 65536 bytes" in answer +def test_answers_400_to_a_content_length_with_too_many_digits(start: Any) -> None: + worker, _ = start(Held()) + + for expect in ("", "Expect: 100-continue\r\n"): + answer = raw_request( + worker, + ( + f"POST {RUN} HTTP/1.1\r\nHost: localhost\r\n{expect}" + f"Content-Length: {'9' * 5000}\r\n\r\n" + ).encode(), + ) + + # With Expect, the length is not known to be over the limit. So the + # worker sends 100 Continue, and then the 400 for the invalid header. + assert b"HTTP/1.1 400" in answer, expect + assert answer.endswith(b'{"error":"request Content-Length is not valid"}') + + def test_answers_expect_100_continue_before_it_reads_the_body(start: Any) -> None: worker, client = start(routes={"/build": lambda job_input, context: "ok"}) data = json.dumps(job_body()).encode() @@ -1319,6 +1337,36 @@ def failing_start(self: threading.Thread) -> None: assert "could not start the run hook job" in logger.messages("error") +def test_fails_a_job_whose_heartbeats_cannot_start( + start: Any, logger: Any, monkeypatch: pytest.MonkeyPatch +) -> None: + handler = Held() + handler.release.set() + worker, client = start(routes={"/build": handler}) + original = threading.Thread.start + + def failing_start(self: threading.Thread) -> None: + if self.name.startswith("heartbeats-"): + msg = "can't start new thread" + raise RuntimeError(msg) + original(self) + + monkeypatch.setattr(threading.Thread, "start", failing_start) + # The job already has its 202, so no delivery follows. The worker fails + # the callback, instead of leaving the job to its heartbeat timeout. + assert request(worker, "/build", job_body(heartbeatTimeoutSeconds=60))[0] == 202 + assert worker.idle(WAIT) + + [(name, kwargs)] = completions(client) + assert name == "failure" + assert kwargs["Error"] == { + "ErrorType": "RuntimeError", + "ErrorMessage": "can't start new thread", + } + assert handler.inputs == [] + assert "could not start the job's heartbeats" in logger.messages("error") + + def test_counts_no_report_whose_thread_cannot_start( start: Any, monkeypatch: pytest.MonkeyPatch ) -> None: