diff --git a/packages/aws-durable-execution-sdk-python-extras/README.md b/packages/aws-durable-execution-sdk-python-extras/README.md index d064a14d3..f66c5ac64 100644 --- a/packages/aws-durable-execution-sdk-python-extras/README.md +++ b/packages/aws-durable-execution-sdk-python-extras/README.md @@ -123,7 +123,7 @@ except MicrovmError: ... # Any other MicroVM failure. ``` -The class is the same on the first run and on every replay, so code that branches on it stays deterministic. A `microvm` call inside your own child context, or inside a `map` or `parallel` item, reaches the code around that context as the SDK's `ChildContextError`. Its `__cause__` is the MicroVM error, so check `isinstance(error.__cause__, MicrovmTimeoutError)` there. An invalid name or config raises the SDK's `ValidationError`, and an input that is not JSON-serializable raises `TypeError`, before any durable operation starts. `Duration` raises the same `ValidationError` for a negative value. The operation does not check `image_identifier` and `execution_role_arn` before it starts. boto3 checks them against the API model before it sends `RunMicrovm`, and the service checks them again. So an invalid value fails the launch step at once with `MicrovmLaunchError`, after the callback step. +The class is the same on the first run and on every replay, so code that branches on it stays deterministic. A `microvm` call inside your own child context, or inside a `map` or `parallel` item, reaches the code around that context as the SDK's `ChildContextError`. Its `__cause__` is the MicroVM error, so check `isinstance(error.__cause__, MicrovmTimeoutError)` there. The operation raises three errors before any durable operation starts. An invalid name or config raises the SDK's `ValidationError`. An input that is not JSON-serializable raises `TypeError`. An input over `MAX_JOB_INPUT_BYTES` raises `ValueError`. `Duration` raises the same `ValidationError` for a negative value. The operation does not check `image_identifier` and `execution_role_arn` before it starts. boto3 checks them against the API model before it sends `RunMicrovm`, and the service checks them again. So an invalid value fails the launch step at once with `MicrovmLaunchError`, after the callback step. A failed AWS call keeps its request ID in the error message, in the form ` request ID: .`, for example `RunMicrovm request ID: .`. A support case can quote it. This applies to `RunMicrovm`, `GetMicrovm`, `ResumeMicrovm`, `CreateMicrovmAuthToken`, and `TerminateMicrovm`. The SDK checkpoints the message of a failed step, and not its `__cause__`. So the request ID is in the message, the handler sees it on the first run and on replay, and the execution history keeps it. A failed `TerminateMicrovm` call does not fail the operation. So a `TerminateMicrovm` request ID appears only in the terminate log entry and in the terminate step record. A message has no request ID when the answer carries none, for example after a connection failure. A message also has no request ID when the error is not the answer of the Lambda MicroVMs call. For example, a `client` with role-assuming credentials calls STS `AssumeRole` inside each call, and an STS error carries the request ID of STS. The operation recognizes such an error by its `operation_name`. The step raises a copy of the botocore error, of the same class, with the same `response`, and with the request ID in its message. So `default_microvm_retry_strategy` decides as before. A custom `retry_strategy` receives the error that botocore raised, without the request ID. So a strategy that matches the message, such as `create_retry_strategy(RetryStrategyConfig(retryable_errors=[...]))`, decides on the message of the service. A `ResourceNotFoundException` from `GetMicrovm`, `ResumeMicrovm`, or `CreateMicrovmAuthToken` is the exception. It means that the MicroVM no longer exists. So the operation raises `MicrovmStateError` instead, and the strategy receives that error. Its `__cause__` is the botocore error. An error that the operation creates itself, such as `MicrovmStateError` or "RunMicrovm returned no microvmId", has the request ID in its message when the strategy receives it. A custom client can raise one error object for several calls. Each step message then names only its own call, and the object keeps its message. @@ -146,6 +146,7 @@ The 10-second reserve keeps time for the step to record its failure. A Lambda ti - `timeout` must be between 1 second and 8 hours, because a MicroVM lives at most 8 hours. The MicroVM's `maximumDurationInSeconds` is `timeout` plus 5 minutes, at most 8 hours. So a timeout above 7 hours 55 minutes leaves the MicroVM less than 5 minutes beyond the callback's timeout. - `job_input` is sent as JSON. So the MicroVM receives what `json.dumps` writes: a tuple arrives as a list, and a dict key that is not a string arrives as a string. +- The worker accepts a job request body of at most 6 MiB. The body holds the JSON of `job_input` and a few job fields. The job fields take less than 3 KiB. So `job_input` can be at most `MAX_JOB_INPUT_BYTES`, 6 MiB minus 8 KiB, as UTF-8 JSON. A larger input raises `ValueError` from `microvm` or `vm.invoke` before any durable operation starts. - The MicroVM gets the `INTERNET_EGRESS` connector, because it needs egress to reach the Lambda callback APIs. For ingress, it gets `NO_INGRESS` with `run` hook delivery and `ALL_INGRESS` with HTTP delivery. If you set `ingress_network_connectors`, allow ingress for every job that goes over HTTP. A `microvm` job goes over HTTP when its `run` hook payload is over 4,096 code points, or when its config sets `request.path`. Every session job goes over HTTP. An empty tuple sends no connector. - `heartbeat_timeout` starts when the callback is created, before the launch. The worker sends its first heartbeat when the job arrives. So the first period must cover the launch, the boot, and, for HTTP delivery, the request step, each with its retries. A throttled launch can take several minutes, and the request step retries for up to `retry_window` in each attempt. With a shorter heartbeat timeout, the operation fails with `MicrovmTimeoutError` before the MicroVM sends its first heartbeat. - The request step needs at least 11 seconds left in the invocation when it starts. It keeps a 10-second reserve before the invocation times out, and it starts a call only with at least 1 second left before the reserve. The launch step can run earlier in the same invocation, so the function timeout must cover it too. A function timeout of `retry_window` plus 11 seconds, 71 seconds by default, leaves the first retry tier about its whole window. With less than 11 seconds left, every attempt fails with `MicrovmEndpointUnavailableError`, and the message of the final `MicrovmDeliveryError` names the function timeout. The Lambda default timeout is 3 seconds. With it, `microvm` runs small jobs in the `run` hook and fails every job that goes over HTTP: a job too large for the hook, and a job with `request.path`. @@ -240,7 +241,7 @@ A failed `vm.invoke` raises its MicroVM error inside the handler. The handler ca 1. A MicroVM error that the handler raises itself reaches the caller with its own class, such as `MicrovmJobFailedError`. 2. Any other error reaches the caller as the SDK's `ChildContextError`, as from `run_in_child_context`. This includes a MicroVM error that leaves a child context that the handler created: that child context records `ChildContextError` as its error type. The caller then gets the MicroVM error's message, but not its class. Inside the handler, the `ChildContextError` of that child context has the MicroVM error as its `__cause__`. An SDK error that leaves the handler, such as the `StepError` of a failed step, also reaches the caller as `ChildContextError`, with the SDK error as its `__cause__`. This rule matches `run_in_child_context`. The JavaScript operation differs here: it passes an SDK error to the caller with its own class. -An invalid job raises `ValidationError` or `TypeError` from `vm.invoke`, before the job's durable operations start. The handler can catch it too. A call without the context, `vm.invoke("build", job_input=..., options=...)`, raises Python's `TypeError`. A first argument that is not a durable context, as in `vm.invoke("build", "build", ...)`, raises `AttributeError`. +An invalid job raises `ValidationError`, `TypeError`, or `ValueError` from `vm.invoke`, before the job's durable operations start. The handler can catch it too. A call without the context, `vm.invoke("build", job_input=..., options=...)`, raises Python's `TypeError`. A first argument that is not a durable context, as in `vm.invoke("build", "build", ...)`, raises `AttributeError`. The handler's return value is checkpointed as the session's result. So it must be serializable by the SDK, and it counts toward the checkpoint size limit. @@ -299,9 +300,9 @@ The MicroVM operations handle the network, the service, and the worker lifecycle ### Job inputs 1. Pass a JSON-serializable input with finite numbers. Any other input raises `TypeError` before any durable operation starts. See [Errors](#errors). -2. Keep the serialized job request at or under 6 MiB. The job request holds the input and a few job fields. Pass larger data by reference, for example as an S3 key. - - The Python worker reads a job request body of at most 6 MiB. It answers a larger body with HTTP 400. - - The operation does not check the size before it sends the job. So a larger body reaches the worker, and the job fails with `MicrovmDeliveryError`. It fails at once when the 400 reaches the operation. It fails only after the retries of both tiers when the connection is reset instead, and the MicroVM is billed meanwhile. [What the worker does](../aws-durable-execution-sdk-python-microvm-worker/README.md#what-the-worker-does) says when each case happens. +2. Keep the job input's JSON at or under `MAX_JOB_INPUT_BYTES`, 6 MiB minus 8 KiB, in UTF-8. Pass larger data by reference, for example as an S3 key. + - The Python worker reads a job request body of at most 6 MiB. The body holds the input's JSON and a few job fields. + - `microvm` and `vm.invoke` measure the input's JSON before any durable operation starts. A larger input raises `ValueError`. So `microvm` launches no MicroVM for it. 3. Expect a job to go over HTTP in three cases: - The `run` hook payload of a `microvm` job is over 4,096 code points. The payload holds the input, the callback ID, and a few other fields. - The `microvm` config sets `request.path`. The job then goes over HTTP for any input size. diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/__init__.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/__init__.py index 1c0c04634..9d3095761 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/__init__.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/__init__.py @@ -25,6 +25,7 @@ MicrovmTimeoutError, ) from aws_durable_execution_sdk_python_extras.microvm.job import ( + MAX_JOB_INPUT_BYTES, MAX_RUN_HOOK_PAYLOAD_LENGTH, ) from aws_durable_execution_sdk_python_extras.microvm.operation import microvm @@ -49,6 +50,7 @@ "DEFAULT_AUTO_SUSPEND_IDLE_SECONDS", "DEFAULT_MICROVM_JOB_PATH", "DEFAULT_MICROVM_PORT", + "MAX_JOB_INPUT_BYTES", "MAX_MICROVM_DURATION_SECONDS", "MAX_RUN_HOOK_PAYLOAD_LENGTH", "MicrovmBaseConfig", diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/job.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/job.py index 9cc7979d6..59e7bee0e 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/job.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/job.py @@ -29,6 +29,27 @@ counts what the service counts. """ +MAX_JOB_INPUT_BYTES = 6 * 1024 * 1024 - 8 * 1024 +"""The largest job input, in UTF-8 bytes of its JSON, that ``microvm`` and +``vm.invoke`` accept. + +1. The worker accepts a job request body of at most 6 MiB. +2. The body holds the input as the same JSON that :func:`dumps` writes. +3. The body also holds fixed fields: the version, the Region, the MicroVM + ID, the callback ID, and the heartbeat timeout. Their names and the JSON + punctuation take 92 bytes. +4. The botocore ``lambda-microvms`` model allows a MicroVM ID of 1 to 256 + characters, with no pattern. JSON writes a character in at most 6 bytes, + as a ``\\u00XX`` escape. So the MicroVM ID takes at most 1,536 bytes. +5. The Lambda model allows a callback ID of at most 1,024 characters of + ``[A-Za-z0-9+/]+={0,2}``. JSON writes each of these characters in 1 byte. + So the callback ID takes at most 1,024 bytes. +6. The Region is short ASCII. The heartbeat timeout is a few digits. +7. So the fixed fields take less than 3 KiB. So 8 KiB covers them. + +So an input whose JSON has at most 6 MiB minus 8 KiB fits in the body. +""" + @dataclass(frozen=True, kw_only=True) class RunHookPayload: diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py index c4d3ea67f..b7e5759ee 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/operation.py @@ -6,7 +6,6 @@ from __future__ import annotations import hashlib -import json import os import re import threading @@ -57,8 +56,10 @@ MicrovmTimeoutError, ) from aws_durable_execution_sdk_python_extras.microvm.job import ( + MAX_JOB_INPUT_BYTES, JobRequest, RunHookPayload, + dumps, fits_run_hook, run_hook_payload_without_job, ) @@ -258,6 +259,9 @@ def microvm( value. TypeError: When ``job_input`` is not JSON-serializable. No durable operation starts. + ValueError: When the JSON of ``job_input`` is over + :data:`MAX_JOB_INPUT_BYTES` UTF-8 bytes. No durable operation + starts. MicrovmLaunchError: When ``RunMicrovm`` fails after all retries, or rejects the config, for example an empty ``image_identifier``. MicrovmNotRunningError: When ``CreateMicrovmAuthToken`` reports that @@ -356,16 +360,29 @@ def _validate_heartbeat(name: str, heartbeat_timeout: Duration | None) -> int | def _validate_input(name: str, job_input: Any) -> None: - """Check that the job input is JSON-serializable. + """Check that the job input is JSON-serializable, and fits in a job body. + + The size is the UTF-8 length of the JSON that the job body holds. So it + is measured on the output of :func:`dumps`. Raises: - TypeError: When it is not. + TypeError: When it is not JSON-serializable. + ValueError: When its JSON is over :data:`MAX_JOB_INPUT_BYTES` bytes. """ try: - json.dumps(job_input, allow_nan=False) + text = dumps(job_input) except (TypeError, ValueError) as error: msg = f'MicroVM "{name}": the job input must be JSON-serializable: {error}' raise TypeError(msg) from error + size = len(text.encode()) + if size > MAX_JOB_INPUT_BYTES: + msg = ( + f'MicroVM "{name}": the job input is {size} bytes as JSON. The limit ' + f"is {MAX_JOB_INPUT_BYTES} bytes, because the worker accepts a job " + "body of at most 6 MiB. Pass larger data by reference, for example " + "as an S3 key." + ) + raise ValueError(msg) def _validate_region(context: DurableContext, name: str) -> tuple[str, str]: @@ -720,6 +737,7 @@ def create( ValidationError: When ``name`` or ``config`` is not valid, or the Region cannot be determined. TypeError: When ``job_input`` is not JSON-serializable. + ValueError: When the JSON of ``job_input`` is too large. """ timeout_seconds = _validate_base(name, config.timeout) heartbeat_timeout_seconds = _validate_heartbeat(name, config.heartbeat_timeout) diff --git a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py index 4f3427224..6cd824c0a 100644 --- a/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py +++ b/packages/aws-durable-execution-sdk-python-extras/src/aws_durable_execution_sdk_python_extras/microvm/session.py @@ -250,6 +250,8 @@ def invoke( AttributeError: When ``context`` is not a durable context, for example a job name passed as the first argument. Python raises it at ``context.run_in_child_context``. + ValueError: When the JSON of ``job_input`` is over + ``MAX_JOB_INPUT_BYTES`` UTF-8 bytes. ValidationError: When ``name`` or ``options`` is not valid. MicrovmNotRunningError: When the session's MicroVM has ended. The session reports it when it checks the MicroVM state before diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py index ff302d639..0045dcd4c 100644 --- a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/operation_test.py @@ -36,6 +36,7 @@ from botocore.stub import ANY, Stubber from aws_durable_execution_sdk_python_extras.microvm import ( + MAX_JOB_INPUT_BYTES, MAX_RUN_HOOK_PAYLOAD_LENGTH, MicrovmConfig, MicrovmError, @@ -366,6 +367,33 @@ def run_microvm(self, **request: Any) -> dict[str, Any]: assert "RunMicrovm returned no microvmId" in error["message"] +def test_rejects_a_job_input_over_the_size_limit_before_the_launch( + run: Any, fake_client: Any +) -> None: + client = fake_client() + + @durable_execution + def handler(event: Any, context: DurableContext) -> Any: + # The JSON of the input is the limit plus 1 byte, with the 2 quotes. + job_input = "x" * (MAX_JOB_INPUT_BYTES - 1) + try: + microvm(context, "build", job_input=job_input, config=config_with(client)) + except ValueError as error: + return str(error) + return None + + result = run(handler, {}) + + assert outcome(result) == ( + f'MicroVM "build": the job input is {MAX_JOB_INPUT_BYTES + 1} bytes as ' + f"JSON. The limit is {MAX_JOB_INPUT_BYTES} bytes, because the worker " + "accepts a job body of at most 6 MiB. Pass larger data by reference, " + "for example as an S3 key." + ) + assert client.calls == [] + assert result.operations == [] + + def test_sends_a_job_too_large_for_the_run_hook_over_http( run: Any, fake_client: Any, fake_endpoint: Any ) -> None: diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py index 78cdfa883..d09f1fa84 100644 --- a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/session_test.py @@ -38,6 +38,7 @@ from aws_durable_execution_sdk_python_extras.microvm import ( DEFAULT_MICROVM_JOB_PATH, + MAX_JOB_INPUT_BYTES, MicrovmError, MicrovmInvokeOptions, MicrovmJobFailedError, @@ -405,6 +406,68 @@ def body(vm: MicrovmSession, ctx: DurableContext, event: Any) -> Any: ] +def test_rejects_a_job_input_over_the_size_limit_before_the_job_starts( + run: Any, fake_client: Any, fake_endpoint: Any +) -> None: + endpoint = fake_endpoint() + client = fake_client(endpoint=endpoint.url) + + def body(vm: MicrovmSession, ctx: DurableContext, event: Any) -> Any: + # The JSON of the input is the limit plus 1 byte, with the 2 quotes. + job_input = "x" * (MAX_JOB_INPUT_BYTES - 1) + try: + vm.invoke(ctx, "build", job_input=job_input, options=job()) + except ValueError as error: + return str(error) + return None + + result = run(session_handler(session_config(client), body)) + + assert outcome(result) == { + "result": ( + f'MicroVM "build": the job input is {MAX_JOB_INPUT_BYTES + 1} bytes ' + f"as JSON. The limit is {MAX_JOB_INPUT_BYTES} bytes, because the " + "worker accepts a job body of at most 6 MiB. Pass larger data by " + "reference, for example as an S3 key." + ) + } + # Only the session's own launch ran. No job operation started, and no + # request reached the MicroVM. + assert len(client.requests("run_microvm")) == 1 + assert client.requests("create_microvm_auth_token") == [] + assert endpoint.requests == [] + session = result.get_context("pipeline") + assert [operation.name for operation in session.child_operations] == [ + "pipeline.session", + "pipeline.launch", + "pipeline.terminate", + ] + + +def test_sends_a_job_input_at_the_size_limit( + run: Any, fake_client: Any, fake_endpoint: Any +) -> None: + endpoint = fake_endpoint() + client = fake_client(endpoint=endpoint.url) + # The JSON of the input is the limit, with the 2 quotes. Each "é" takes + # 2 bytes. + job_input = "é" * ((MAX_JOB_INPUT_BYTES - 2) // 2) + + def body(vm: MicrovmSession, ctx: DurableContext, event: Any) -> Any: + return vm.invoke(ctx, "build", job_input=job_input, options=job()) + + result = run( + session_handler(session_config(client), body), + None, + [("build", succeed_with("done"))], + ) + + assert outcome(result) == {"result": "done"} + [posted] = endpoint.requests + assert posted["body"]["input"] == job_input + assert int(posted["headers"]["content-length"]) <= 6 * 1024 * 1024 + + def test_retries_a_throttled_launch_with_the_same_client_token( run: Any, fake_client: Any, fake_endpoint: Any, make_error: Any ) -> None: diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/validation_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/validation_test.py index 2dbf3efa8..8623f2a00 100644 --- a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/validation_test.py +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/validation_test.py @@ -13,6 +13,7 @@ from aws_durable_execution_sdk_python.exceptions import ValidationError from aws_durable_execution_sdk_python_extras.microvm import ( + MAX_JOB_INPUT_BYTES, MicrovmConfig, MicrovmRequestConfig, microvm, @@ -187,6 +188,69 @@ def test_rejects_an_input_that_is_not_json(job_input: Any) -> None: microvm(NoOperations(), "build", job_input=job_input, config=config()) # type: ignore[arg-type] +class StartsOperations(NoOperations): + """A context that records the start of the operation's child context, + and runs nothing in it.""" + + def __init__(self) -> None: + super().__init__() + self.started: list[str] = [] + + def run_in_child_context(self, func: Any, name: str, **kwargs: Any) -> Any: + self.started.append(name) + return "started" + + +def input_of_size(character: str, width: int, size: int) -> str: + """A string whose JSON is ``size`` UTF-8 bytes. + + ``character`` takes ``width`` bytes in the JSON. The two quotes take two. + Single-byte padding fills the rest. + """ + count = (size - 2) // width + return character * count + "x" * (size - 2 - count * width) + + +def too_large(size: int) -> str: + return ( + f'MicroVM "build": the job input is {size} bytes as JSON. The limit is ' + "6283264 bytes, because the worker accepts a job body of at most 6 MiB. " + "Pass larger data by reference, for example as an S3 key." + ) + + +def test_limits_the_job_input_to_6_mib_minus_8_kib() -> None: + assert MAX_JOB_INPUT_BYTES == 6 * 1024 * 1024 - 8 * 1024 == 6_283_264 + + +# Each character takes its UTF-8 length in the JSON. A lone surrogate takes +# the 6 bytes of its \uXXXX escape, because that is how the job body holds it. +@pytest.mark.parametrize( + ("character", "width"), + [("x", 1), ("é", 2), ("日", 3), ("😀", 4), ("\ud800", 6)], + ids=["ascii", "two-byte", "three-byte", "four-byte", "lone-surrogate"], +) +def test_counts_the_job_input_in_utf8_bytes_of_its_json( + character: str, width: int +) -> None: + at_limit = input_of_size(character, width, MAX_JOB_INPUT_BYTES) + context = StartsOperations() + + assert ( + microvm(context, "build", job_input=at_limit, config=config()) # type: ignore[arg-type] + == "started" + ) + assert context.started == ["build"] + + over_limit = at_limit + "x" + with pytest.raises(ValueError, match="bytes as JSON") as raised: + microvm(NoOperations(), "build", job_input=over_limit, config=config()) # type: ignore[arg-type] + assert str(raised.value) == too_large(MAX_JOB_INPUT_BYTES + 1) + if width > 1: + # Counting code points would accept the input. + assert len(over_limit) + 2 < MAX_JOB_INPUT_BYTES + + def test_rejects_a_call_without_a_region(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("AWS_REGION") diff --git a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/worker_composed_test.py b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/worker_composed_test.py index 5fa7dfe6a..06aaf0464 100644 --- a/packages/aws-durable-execution-sdk-python-extras/tests/microvm/worker_composed_test.py +++ b/packages/aws-durable-execution-sdk-python-extras/tests/microvm/worker_composed_test.py @@ -44,6 +44,7 @@ from aws_durable_execution_sdk_python_testing.runner import DurableFunctionTestRunner from aws_durable_execution_sdk_python_extras.microvm import ( + MAX_JOB_INPUT_BYTES, MAX_RUN_HOOK_PAYLOAD_LENGTH, MicrovmConfig, MicrovmError, @@ -302,6 +303,20 @@ def job(job_input: Any, context: MicrovmJobContext) -> Any: assert client.requests("terminate_microvm") == [{"microvmIdentifier": "mvm-1"}] +def test_runs_a_job_input_at_the_size_limit_in_the_worker(composed: Any) -> None: + # The job body adds the job fields to the input. The worker accepts the + # body, so the fields fit in the 8 KiB that the limit leaves. + def job(job_input: Any, context: MicrovmJobContext) -> Any: + return len(job_input) + + # The JSON of the input is the limit, with the 2 quotes. + outcome, _, _ = composed( + job, "x" * (MAX_JOB_INPUT_BYTES - 2), heartbeat_timeout=Duration.from_hours(8) + ) + + assert outcome == {"result": MAX_JOB_INPUT_BYTES - 2} + + def test_runs_a_job_on_a_route_of_the_worker(composed: Any) -> None: def job(job_input: Any, context: MicrovmJobContext) -> Any: return "handler" 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 209b6ead6..851063f90 100644 --- a/packages/aws-durable-execution-sdk-python-microvm-worker/README.md +++ b/packages/aws-durable-execution-sdk-python-microvm-worker/README.md @@ -96,7 +96,7 @@ A second delivery of a job is answered like the first and ignored, while the job The worker creates a Lambda client for each job, after the job arrives, not at image build. A client created before the snapshot would carry build-time state into every MicroVM. A client kept across jobs would keep connections that a suspend of the MicroVM can leave dead. The worker closes each job's default client when the job ends. It never closes a client from `create_client`. -A document that does not match the contract gets HTTP 400 with the validation message. If the document named a callback, the worker also fails that callback, so the durable function fails at once instead of at its timeout. A request whose body cannot be read gets HTTP 400 with a generic message, and the error is logged. The worker accepts a `run` hook body of at most 64 KB and a job request body of at most 6 MB, with a `Content-Length` or with chunked transfer encoding. In a chunked body, the chunk-size lines, their extensions, and the trailers have a limit of their own, of the same size. A larger body gets HTTP 400. With `Expect: 100-continue`, the 400 replaces `100 Continue`, so the client does not send the body. Without it, the worker reads and drops the body after the answer, for at most 10 seconds and 24 MB. Without that read, the connection would close with unread bytes, and the client would get a TCP reset instead of the 400. A body that is larger, or that is still arriving after 10 seconds, still gets the reset. Other lifecycle hooks get HTTP 200. Other paths get HTTP 404. +A document that does not match the contract gets HTTP 400 with the validation message. If the document named a callback, the worker also fails that callback, so the durable function fails at once instead of at its timeout. A request whose body cannot be read gets HTTP 400 with a generic message, and the error is logged. The worker accepts a `run` hook body of at most 64 KiB and a job request body of at most 6 MiB, with a `Content-Length` or with chunked transfer encoding. In a chunked body, the chunk-size lines, their extensions, and the trailers have a limit of their own, of the same size. A larger body gets HTTP 400. The worker sends it without reading the body, and then closes the connection. The unread bytes make the kernel send a TCP reset in place of a normal close. So a client that is still sending can get the reset before it reads the 400. A request with `Expect: 100-continue` first gets `100 Continue`, as `http.server` sends it. The client then sends the body, and gets the 400 or the reset in the same way. The MicroVM operations of both SDK extras packages, JS (`@aws/durable-execution-sdk-js-extras`) and Python (`aws-durable-execution-sdk-python-extras`), reject a job input that does not fit before they send the job. So only another sender can get the reset. Other lifecycle hooks get HTTP 200. Other paths get HTTP 404. `worker.idle(timeout)` waits until every job started so far has ended, and a running `SuspendMicrovm` call has returned. It returns `False` when the timeout ends first. `worker.close()` stops the server and the idle time. Running jobs continue, so call `idle()` first to let them report their outcome. 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 f07615675..316d2e89f 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 @@ -19,7 +19,6 @@ import socketserver import sys import threading -import time from collections import OrderedDict from collections.abc import Callable, Mapping from dataclasses import dataclass @@ -120,14 +119,6 @@ _MAX_CHUNK_LINE_BYTES = 1024 -_DISCARD_SECONDS = 10.0 -"""How long the worker reads and drops a body that it answered with 400.""" - -_MAX_DISCARD_BYTES = 4 * MAX_JOB_BODY_BYTES -"""How many bytes of such a body the worker reads and drops at most.""" - -_DISCARD_READ_BYTES = 64 * 1024 - TERMINATE_REPORT_BUDGET_SECONDS = 5.0 """How long the worker holds the ``terminate`` hook open while it fails the callbacks of running jobs. @@ -589,9 +580,6 @@ class _Handler(_RequestHandler): def handle_exchange(self) -> None: listener.handle_request(self) - def handle_expect_100(self) -> bool: - return listener.handle_expect_100(self) - return _Handler def handle_request(self, request: BaseHTTPRequestHandler) -> None: @@ -609,31 +597,6 @@ def handle_request(self, request: BaseHTTPRequestHandler) -> None: ) if not exchange.responded: exchange.respond(500, {"error": "internal error"}) - finally: - exchange.discard_unread_body() - - def handle_expect_100(self, request: BaseHTTPRequestHandler) -> bool: - """Answer a request that sent ``Expect: 100-continue``. - - :attr:`request_handler_class` calls it before it reads the body. - - 1. A request whose ``Content-Length`` is over the path's limit gets - its HTTP 400 now, instead of ``100 Continue``. RFC 9110 section - 10.1.1 allows a final answer in place of ``100 Continue``. The - client then does not send the body. - 2. Every other request gets ``100 Continue``. - - Returns: - Whether the base class goes on to read the request. - """ - exchange = _Exchange(request, self._logger) - limit = self._body_limit(exchange) - length = exchange.declared_length() - if limit is not None and length is not None and length > limit: - self._reject(exchange.too_large(limit), exchange) - exchange.discard_unread_body() - return False - return BaseHTTPRequestHandler.handle_expect_100(request) def idle(self, timeout: float | None = None) -> bool: """Wait until every job started so far has ended. @@ -828,19 +791,6 @@ def _job_handler(self, path: str) -> MicrovmJobHandler | None: """The handler for a job route, or ``None`` for another path.""" return self._handler if path == MICROVM_JOB_PATH else self._routes.get(path) - def _body_limit(self, exchange: _Exchange) -> int | None: - """The largest body that the request's path reads. - - ``None`` for a request whose body the worker does not read. - """ - if exchange.method != "POST": - return None - if exchange.path == f"{HOOK_PATH_PREFIX}run": - return MAX_HOOK_BODY_BYTES - if self._job_handler(exchange.path) is not None: - return MAX_JOB_BODY_BYTES - return None - def _handle_run_hook(self, exchange: _Exchange) -> None: try: parsed = RunHookRequest.from_dict(exchange.read_json(MAX_HOOK_BODY_BYTES)) @@ -1916,8 +1866,6 @@ def __init__( self._request = request self._logger = logger self.responded = False - # Set when the worker answers before it has read the whole body. - self._body_unread = False @property def method(self) -> str: @@ -1931,20 +1879,6 @@ def header(self, name: str) -> str | None: """A request header, or ``None``. The name is not case-sensitive.""" return self._request.headers.get(name) - def declared_length(self) -> int | None: - """The body length from ``Content-Length``. - - ``None`` when the body is chunked, or the header is absent or not - valid. Reading the body reports an invalid header. - """ - headers = self._request.headers - if "chunked" in headers.get("Transfer-Encoding", "").lower(): - return None - raw = headers.get("Content-Length") - if raw is None or not _CONTENT_LENGTH.fullmatch(raw.strip()): - return None - return int(raw) - def read_json(self, max_bytes: int) -> Any: """Read the body, and parse it as JSON. @@ -1960,17 +1894,6 @@ def read_json(self, max_bytes: int) -> Any: msg = "request body is not valid JSON" raise InvalidRunHookPayloadError(msg) from error - def too_large( - self, max_bytes: int, part: str = "request body" - ) -> InvalidRunHookPayloadError: - """Return the error for a request part over ``max_bytes``. - - The worker answers such a request before it reads the rest of the - body. So this method also marks the body unread. - """ - self._body_unread = True - return InvalidRunHookPayloadError(f"{part} exceeds {max_bytes} bytes") - def respond(self, status: int, body: Mapping[str, object]) -> None: """Send the answer. A client that has gone away is logged.""" self.responded = True @@ -1990,54 +1913,19 @@ def respond(self, status: int, body: Mapping[str, object]) -> None: extra={"status": status, "error": describe(error)}, ) - def discard_unread_body(self) -> None: - """Read and drop the rest of a body that the worker did not read. - - 1. The worker answers a body over the limit without reading it. - 2. The connection then closes with unread bytes in its receive - buffer. So the kernel sends a TCP reset instead of a normal close. - 3. A client that is still sending gets the reset before it reads the - answer. So the client never sees the 400. - 4. The durable function fails at once on a 400, and retries a reset. - So the reset turns a clear rejection into retries. - - So after the answer, the worker ends its side of the connection, and - reads and drops input until the client closes. The read stops after - :data:`_DISCARD_SECONDS` or :data:`_MAX_DISCARD_BYTES`. A client that - sends more still gets the reset. - """ - if not self._body_unread: - return - connection = self._request.connection - try: - connection.shutdown(socket.SHUT_WR) - deadline = time.monotonic() + _DISCARD_SECONDS - discarded = 0 - while discarded < _MAX_DISCARD_BYTES: - remaining = deadline - time.monotonic() - if remaining <= 0: - return - connection.settimeout(remaining) - data = connection.recv(_DISCARD_READ_BYTES) - if not data: - return - discarded += len(data) - except OSError: - # The client has gone, or the time ran out in recv. - return - def _read_body(self, max_bytes: int) -> bytes: headers = self._request.headers if "chunked" in headers.get("Transfer-Encoding", "").lower(): return self._read_chunked(max_bytes) - if headers.get("Content-Length") is None: + raw = headers.get("Content-Length") + if raw is None: return b"" - length = self.declared_length() - if length is None: + if not _CONTENT_LENGTH.fullmatch(raw.strip()): msg = "request Content-Length is not valid" raise InvalidRunHookPayloadError(msg) + length = int(raw) if length > max_bytes: - raise self.too_large(max_bytes) + raise _too_large(max_bytes) return self._read_exactly(length) def _read_chunked(self, max_bytes: int) -> bytes: @@ -2066,9 +1954,7 @@ def read_framing_line() -> bytes: line = self._request.rfile.readline(_MAX_CHUNK_LINE_BYTES) framing += len(line) if framing > max_bytes: - raise self.too_large( - max_bytes, "the chunked framing of the request body" - ) + raise _too_large(max_bytes, "the chunked framing of the request body") return line while True: @@ -2086,7 +1972,7 @@ def read_framing_line() -> bytes: return b"".join(chunks) size += chunk_size if size > max_bytes: - raise self.too_large(max_bytes) + raise _too_large(max_bytes) chunks.append(self._read_exactly(chunk_size)) if self._read_exactly(2) != b"\r\n": raise _malformed_chunks() @@ -2098,6 +1984,28 @@ def _read_exactly(self, length: int) -> bytes: return data +def _too_large( + max_bytes: int, part: str = "request body" +) -> InvalidRunHookPayloadError: + """Return the error for a request part over ``max_bytes``. + + 1. The worker answers HTTP 400 without reading the rest of the body. + 2. The answer closes the connection. So the unread bytes are never + parsed as the next request. + 3. The connection closes with unread bytes. So the kernel sends a TCP + reset in place of a normal close. + 4. A client that is still sending can get the reset before it reads + the 400. + + The MicroVM operations of both SDK extras packages, JS + (``@aws/durable-execution-sdk-js-extras``) and Python + (``aws-durable-execution-sdk-python-extras``), reject a job input that + does not fit before they send the job. So only another sender can get + the reset. + """ + return InvalidRunHookPayloadError(f"{part} exceeds {max_bytes} bytes") + + def _malformed_chunks() -> InvalidRunHookPayloadError: return InvalidRunHookPayloadError("request body has malformed chunked encoding") 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 1d97b0380..4e0cd2605 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 @@ -5,6 +5,7 @@ from __future__ import annotations +import contextlib import functools import http.client import io @@ -157,13 +158,21 @@ def request( def raw_request(worker: MicrovmWorker, data: bytes) -> bytes: - """Send bytes, close the sending side, and return the raw answer.""" + """Send bytes, close the sending side, and return the raw answer. + + The worker answers some requests without reading all of the bytes. The + connection then closes with unread bytes. So the kernel sends a TCP + reset, and a write or a later read fails. The answer arrives before the + reset. So this returns the bytes read before any failure. + """ + chunks = [] with socket.create_connection((host_of(worker), worker.port), timeout=WAIT) as sock: - sock.sendall(data) - sock.shutdown(socket.SHUT_WR) - chunks = [] - while chunk := sock.recv(65536): - chunks.append(chunk) + with contextlib.suppress(ConnectionError): + sock.sendall(data) + sock.shutdown(socket.SHUT_WR) + with contextlib.suppress(ConnectionError): + while chunk := sock.recv(65536): + chunks.append(chunk) return b"".join(chunks) @@ -662,8 +671,8 @@ def test_answers_400_to_a_content_length_with_too_many_digits(start: Any) -> Non ).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. + # With Expect, the base class sends 100 Continue before the worker + # sees the request. The worker then sends the 400 for the header. assert b"HTTP/1.1 400" in answer, expect assert answer.endswith(b'{"error":"request Content-Length is not valid"}') @@ -691,31 +700,45 @@ def test_answers_expect_100_continue_before_it_reads_the_body(start: Any) -> Non assert client.names() == ["success"] -def test_answers_400_to_a_body_over_the_limit_that_the_client_sends( - start: Any, +@pytest.mark.parametrize("expect", [False, True]) +def test_answers_400_or_resets_a_body_over_the_limit_that_the_client_sends( + start: Any, logger: Any, expect: bool ) -> None: worker, client = start(routes={"/build": Held()}) - data = b"x" * (worker_module.MAX_JOB_BODY_BYTES + 1) + size = worker_module.MAX_JOB_BODY_BYTES + 1 + head = f"POST /build HTTP/1.1\r\nHost: localhost\r\nContent-Length: {size}\r\n" + if expect: + head += "Expect: 100-continue\r\n" + answer = b"" - status, body = request(worker, "/build", raw=data) + # A hang raises TimeoutError, which the test does not suppress. + with socket.create_connection((host_of(worker), worker.port), timeout=WAIT) as sock: + sock.sendall(f"{head}\r\n".encode()) + if expect: + answer += sock.recv(65536) + # The worker answers 400 and closes the connection without reading + # the body. So the kernel sends a TCP reset, and the client can get + # it before or after it reads the 400. + with contextlib.suppress(ConnectionError): + sock.sendall(b"x" * size) + with contextlib.suppress(ConnectionError): + while chunk := sock.recv(65536): + answer += chunk + assert worker.idle(WAIT) - assert (status, body) == (400, {"error": "request body exceeds 6291456 bytes"}) + statuses = [ + line.split(b" ")[1] + for line in answer.split(b"\r\n") + if line.startswith(b"HTTP/1.1 ") + ] + interim = [b"100"] if expect else [] + assert statuses in (interim, [*interim, b"400"]) + if statuses[-1:] == [b"400"]: + assert answer.endswith(b'{"error":"request body exceeds 6291456 bytes"}') + assert "invalid request" in logger.messages("error") assert client.calls == [] -def test_answers_400_to_a_chunked_body_over_the_limit_that_the_client_sends( - start: Any, -) -> None: - worker, _ = start(Held()) - data = b"x" * (worker_module.MAX_HOOK_BODY_BYTES + 1) - - status, body = request( - worker, RUN, raw=iter([data[:1000], data[1000:]]), chunked=True - ) - - assert (status, body) == (400, {"error": "request body exceeds 65536 bytes"}) - - @pytest.mark.parametrize( ("path", "limit"), [ @@ -723,163 +746,23 @@ def test_answers_400_to_a_chunked_body_over_the_limit_that_the_client_sends( (RUN, worker_module.MAX_HOOK_BODY_BYTES), ], ) -def test_answers_400_instead_of_100_continue_to_a_body_over_the_limit( +def test_answers_100_continue_and_then_400_to_a_body_over_the_limit( start: Any, logger: Any, path: str, limit: int ) -> None: - worker, _ = start(Held(), routes={"/build": Held()}) + # The base class answers "Expect: 100-continue" before the worker sees + # the request. The worker then answers 400 without reading the body. + worker, client = start(Held(), routes={"/build": Held()}) head = ( f"POST {path} HTTP/1.1\r\nHost: localhost\r\nExpect: 100-continue\r\n" f"Content-Length: {limit + 1}\r\n\r\n" ) - with socket.create_connection((host_of(worker), worker.port), timeout=WAIT) as sock: - sock.sendall(head.encode()) - answer = b"" - while chunk := sock.recv(65536): - answer += chunk + answer = raw_request(worker, head.encode()) - assert answer.startswith(b"HTTP/1.1 400") - assert b"100 Continue" not in answer + assert answer.startswith(b"HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 400") assert answer.endswith(f'{{"error":"request body exceeds {limit} bytes"}}'.encode()) assert "invalid request" in logger.messages("error") - - -@pytest.mark.parametrize( - ("method", "path", "framing"), - [ - ("POST", "/other", "Content-Length: 99999999"), - ("POST", f"{HOOK_PATH_PREFIX}ready", "Content-Length: 99999999"), - ("GET", "/build", "Content-Length: 99999999"), - ("POST", "/build", "Transfer-Encoding: chunked"), - ], -) -def test_answers_100_continue_when_the_declared_size_is_not_over_a_limit( - start: Any, method: str, path: str, framing: str -) -> None: - worker, _ = start(Held(), routes={"/build": Held()}) - head = ( - f"{method} {path} HTTP/1.1\r\nHost: localhost\r\nExpect: 100-continue\r\n" - f"{framing}\r\n\r\n" - ) - - with socket.create_connection((host_of(worker), worker.port), timeout=WAIT) as sock: - sock.sendall(head.encode()) - interim = sock.recv(65536) - - assert interim.startswith(b"HTTP/1.1 100 Continue") - - -def _send_after_the_answer( - worker: MicrovmWorker, seconds: float -) -> tuple[bytes, BaseException | None]: - """Get a 400 for an oversized body, then keep sending for ``seconds``. - - The worker shuts down its side right after the answer. So the client - reads the answer and end of file at once. The worker keeps reading and - dropping input until its time limit, and then closes the socket. So a - write after the limit fails. - - Returns: - The answer, and the error of the first failed write, or ``None``. - """ - head = f"POST {RUN} HTTP/1.1\r\nHost: localhost\r\nContent-Length: 999999\r\n\r\n" - with socket.create_connection((host_of(worker), worker.port), timeout=WAIT) as sock: - sock.sendall(head.encode() + b"x" * 100) - answer = b"" - while chunk := sock.recv(65536): - answer += chunk - deadline = time.monotonic() + seconds - try: - while time.monotonic() < deadline: - sock.sendall(b"x" * 4096) - time.sleep(0.01) - except ConnectionError as error: - return answer, error - return answer, None - - -def test_stops_discarding_a_rejected_body_after_the_time_limit( - start: Any, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.setattr(worker_module, "_DISCARD_SECONDS", 0.2) - worker, _ = start(Held()) - - answer, error = _send_after_the_answer(worker, 1.5) - - assert answer.startswith(b"HTTP/1.1 400") - assert isinstance(error, ConnectionError) - - -def test_keeps_discarding_a_rejected_body_until_the_time_limit(start: Any) -> None: - # The default limit is 10 seconds. So writes during the first second must - # succeed. A limit that ended too early would fail them. - worker, _ = start(Held()) - - answer, error = _send_after_the_answer(worker, 1.0) - - assert answer.startswith(b"HTTP/1.1 400") - assert error is None - - -class _EndlessConnection: - """A socket whose peer never stops sending.""" - - def __init__(self) -> None: - self.received = 0 - self.shut = False - - def shutdown(self, how: int) -> None: - self.shut = True - - def settimeout(self, timeout: float) -> None: ... - - def recv(self, size: int) -> bytes: - self.received += size - return b"x" * size - - -def _rejected_exchange(connection: _EndlessConnection, logger: Any) -> Any: - request = type("Request", (), {"connection": connection})() - exchange = worker_module._Exchange(request, logger) # noqa: SLF001 - exchange.too_large(1) - return exchange - - -def test_stops_discarding_a_rejected_body_after_the_byte_limit( - logger: Any, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.setattr(worker_module, "_MAX_DISCARD_BYTES", 200_000) - connection = _EndlessConnection() - - _rejected_exchange(connection, logger).discard_unread_body() - - assert connection.shut - assert 200_000 <= connection.received < 200_000 + 65536 - - -def test_stops_discarding_a_rejected_body_when_the_client_is_gone( - logger: Any, -) -> None: - class GoneConnection(_EndlessConnection): - def shutdown(self, how: int) -> None: - raise ConnectionResetError - - connection = GoneConnection() - - _rejected_exchange(connection, logger).discard_unread_body() - - assert connection.received == 0 - - -def test_stops_discarding_a_rejected_body_when_the_time_is_up( - logger: Any, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.setattr(worker_module, "_DISCARD_SECONDS", 0) - connection = _EndlessConnection() - - _rejected_exchange(connection, logger).discard_unread_body() - - assert connection.received == 0 + assert client.calls == [] def test_accepts_a_chunked_body(start: Any) -> None: @@ -927,6 +810,10 @@ def test_accepts_a_chunked_body_with_extensions_and_trailers(start: Any) -> None (b"2" * 2000, b"malformed chunked encoding"), (b"2\r\n{}\r\n", b'{"error":"invalid request"}'), (b"20000\r\n", b"request body exceeds 65536 bytes"), + ( + b"3e8\r\n" + b"x" * 1000 + b"\r\nfc19\r\n", + b"request body exceeds 65536 bytes", + ), ( b"0\r\n" + b"X: y\r\n" * 20_000 + b"\r\n", _FRAMING_TOO_LARGE, @@ -947,6 +834,7 @@ def test_accepts_a_chunked_body_with_extensions_and_trailers(start: Any) -> None "size-line-too-long", "ends-early", "over-the-limit", + "over-the-limit-in-total", "trailers-over-the-limit", "extensions-over-the-limit", ],