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: