diff --git a/src/backtest_engine/production.py b/src/backtest_engine/production.py index 93601d0..2c3c558 100644 --- a/src/backtest_engine/production.py +++ b/src/backtest_engine/production.py @@ -65,6 +65,14 @@ def _required(environ: Mapping[str, str], name: str) -> str: return value +def _required_uuid(environ: Mapping[str, str], name: str) -> str: + value = _required(environ, name) + try: + return str(uuid.UUID(value)) + except ValueError as exc: + raise ConfigurationError(f"{name} must be a UUID") from exc + + def service_endpoint(environ: Mapping[str, str], service: str) -> str | None: """Resolve an emulator endpoint without coupling S3 and SQS together. @@ -892,6 +900,7 @@ def _feature_object_reader(environ: Mapping[str, str]) -> S3VersionedFeatureObje def orchestrator_job_handler( environ: Mapping[str, str] = os.environ, ) -> OrchestratorJobHandler: + correlation_id = _required_uuid(environ, "BACKTEST_WORKER_CORRELATION_ID") policy = load_runtime_policy(Path(_required(environ, "BACKTEST_RUNTIME_POLICY_FILE"))) engine = _engine(environ) @@ -919,5 +928,5 @@ def orchestrator_job_handler( risk_limits=policy.risk_limits, runtime=BasicPlanRuntime(), wall_clock=lambda: datetime.now(UTC), - correlation_id=_required(environ, "BACKTEST_WORKER_CORRELATION_ID"), + correlation_id=correlation_id, ) diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 00ec5c7..7353956 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -53,6 +53,7 @@ from __future__ import annotations +import logging import uuid from collections.abc import Callable, Iterator, Mapping, Sequence from contextlib import contextmanager @@ -181,6 +182,9 @@ ) +_LOGGER = logging.getLogger(__name__) + + __all__ = [ "API_REQUIRED_ENV", "COST_MODEL_VERSION", @@ -1486,16 +1490,25 @@ def __call__(self, job: Mapping[str, Any], context: JobContext) -> JobOutcome: try: binding = self.bind(envelope, context) except JobNotSatisfiable as exc: - self._publish( - envelope, - self._correlation_id, - status="FAILED", - delivery_attempt=context.receive_count, - failedAt=_utc_text(self._wall_clock()), - attempt=context.attempt_number, - failureCode=exc.reason_code, - retryable=False, - ) + try: + self._publish( + envelope, + self._correlation_id, + status="FAILED", + delivery_attempt=context.receive_count, + failedAt=_utc_text(self._wall_clock()), + attempt=context.attempt_number, + failureCode=exc.reason_code, + retryable=False, + ) + except Exception: + _LOGGER.exception( + "terminal backtest result publish failed run_id=%s attempt=%s reason_code=%s; " + "preserving permanent failure", + envelope.run_id, + context.attempt_number, + exc.reason_code, + ) return JobOutcome(JobResult.PERMANENT_FAILURE, reason_code=exc.reason_code) started_at = self._wall_clock() diff --git a/tests/test_production.py b/tests/test_production.py index 6d67fca..b3a75a8 100644 --- a/tests/test_production.py +++ b/tests/test_production.py @@ -11,6 +11,7 @@ import pytest +import backtest_engine.production as production from backtest_engine.api import RESULT_INGEST_SCOPE from backtest_engine.backtest_request_intake import RequestLane from backtest_engine.production import ( @@ -25,6 +26,7 @@ SqsExecutionJobQueue, api_authenticator, load_execution_policy_catalog, + orchestrator_job_handler, service_endpoint, ) @@ -33,6 +35,18 @@ BOT_ID = UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb") +def test_worker_correlation_id_is_rejected_before_other_worker_dependencies_are_built() -> None: + with pytest.raises(ConfigurationError, match="BACKTEST_WORKER_CORRELATION_ID must be a UUID"): + orchestrator_job_handler({"BACKTEST_WORKER_CORRELATION_ID": "i-07a6870a8c4c199dc"}) + + +def test_worker_correlation_id_is_normalized_to_the_result_event_uuid_format() -> None: + assert production._required_uuid( # type: ignore[attr-defined] + {"BACKTEST_WORKER_CORRELATION_ID": "AAAAAAAA-AAAA-4AAA-8AAA-AAAAAAAAAAAA"}, + "BACKTEST_WORKER_CORRELATION_ID", + ) == "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + + def test_service_specific_aws_endpoint_overrides_the_legacy_shared_endpoint() -> None: environment = { "AWS_ENDPOINT_URL": "http://legacy:4566", diff --git a/tests/test_wiring.py b/tests/test_wiring.py index 71be79a..ceb689b 100644 --- a/tests/test_wiring.py +++ b/tests/test_wiring.py @@ -15,6 +15,8 @@ from __future__ import annotations import copy +import logging +import uuid from datetime import UTC, date, datetime from decimal import ROUND_HALF_UP, Decimal, localcontext from fractions import Fraction @@ -53,6 +55,7 @@ EXECUTION_MODEL_VERSION, BasicPlanReplayFactory, ExecutionModelEngine, + JobEnvelope, JobNotSatisfiable, OrchestratorJobHandler, WiringError, @@ -436,6 +439,88 @@ def publish(self, event: Any, *, delivery_attempt: int) -> None: assert sink.events[0]["deliveryAttempt"] == 3 +def _terminal_failure_envelope() -> JobEnvelope: + return JobEnvelope( + run_id=uuid.UUID("55555555-5555-4555-8555-555555555555"), + bot_id=uuid.UUID("00000000-0000-4000-8000-0000000000b1"), + owner_account_id=uuid.UUID("66666666-6666-4666-8666-666666666666"), + idempotency_key="terminal-publish-failure", + input_bundle_id=uuid.UUID("00000000-0000-4000-8000-0000000000b2"), + input_bundle_fingerprint="sha256:" + "1" * 64, + execution_policy_version="official-backtest-policy-v1", + compiled_plan_checksum="sha256:" + "2" * 64, + dataset_manifest_id=uuid.UUID("00000000-0000-4000-8000-0000000000b3"), + expected_dataset_hash="sha256:" + "3" * 64, + expected_snapshot_hash="sha256:" + "4" * 64, + datasets=(), + feature_materializations=(), + evaluation_period_id=None, + input_set_hash=None, + ) + + +def _terminal_failure_handler() -> OrchestratorJobHandler: + handler = object.__new__(OrchestratorJobHandler) + handler._wall_clock = lambda: COMPLETED_AT + handler._correlation_id = "77777777-7777-4777-8777-777777777777" + + def unsatisfiable(*_args: Any) -> None: + raise JobNotSatisfiable("compiled plan is not resolvable", reason_code="REQUIRED_INPUT_UNAVAILABLE") + + handler.bind = unsatisfiable # type: ignore[method-assign] + return handler + + +def test_unsatisfiable_binding_publishes_the_terminal_failure( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class Sink: + def __init__(self) -> None: + self.events: list[dict[str, Any]] = [] + + def publish(self, event: Any, *, delivery_attempt: int) -> None: + self.events.append(dict(event) | {"deliveryAttempt": delivery_attempt}) + + envelope = _terminal_failure_envelope() + sink = Sink() + handler = _terminal_failure_handler() + handler._sink = sink + context = JobContext("execution-key", 1, 1, "message-1", "worker-1") + monkeypatch.setattr(JobEnvelope, "parse", classmethod(lambda _cls, _job: envelope)) + + outcome = handler({}, context) + + assert outcome.result is JobResult.PERMANENT_FAILURE + assert outcome.reason_code == "REQUIRED_INPUT_UNAVAILABLE" + assert sink.events[0]["status"] == "FAILED" + assert sink.events[0]["failureCode"] == "REQUIRED_INPUT_UNAVAILABLE" + assert sink.events[0]["retryable"] is False + assert sink.events[0]["metadata"]["correlationId"] == "77777777-7777-4777-8777-777777777777" + + +def test_terminal_binding_failure_survives_its_failure_event_publish_failing( + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + envelope = _terminal_failure_envelope() + handler = _terminal_failure_handler() + + def unavailable_sink(*_args: Any, **_kwargs: Any) -> None: + raise OSError("result sink unavailable") + + handler._publish = unavailable_sink # type: ignore[method-assign] + context = JobContext("execution-key", 1, 1, "message-1", "worker-1") + monkeypatch.setattr(JobEnvelope, "parse", classmethod(lambda _cls, _job: envelope)) + + with caplog.at_level(logging.ERROR): + outcome = handler({}, context) + + assert outcome.result is JobResult.PERMANENT_FAILURE + assert outcome.reason_code == "REQUIRED_INPUT_UNAVAILABLE" + assert "result sink unavailable" in caplog.text + assert "REQUIRED_INPUT_UNAVAILABLE" in caplog.text + + # ========================================================================== # The evaluation window comes from the pinned dataset # ==========================================================================