From 9932e40eb39ded2eb8ec59c54f8ede3ccbbe2f56 Mon Sep 17 00:00:00 2001 From: HJ <16863475+hjcud@users.noreply.github.com> Date: Sun, 9 Aug 2026 06:57:04 +0900 Subject: [PATCH] fix: preserve backtest retry reasons --- DEPLOYMENT.md | 6 +++ src/backtest_engine/wiring.py | 16 +++++++- src/backtest_engine/worker.py | 65 ++++++++++++++++++++++++++++++-- tests/test_lane_scheduler.py | 61 +++++++++++++++++++++++++++++- tests/test_wiring_persistence.py | 19 +++++++++- tests/test_worker.py | 34 +++++++++++++++++ 6 files changed, 192 insertions(+), 9 deletions(-) diff --git a/DEPLOYMENT.md b/DEPLOYMENT.md index 71a0cb7..6a00be0 100644 --- a/DEPLOYMENT.md +++ b/DEPLOYMENT.md @@ -113,6 +113,7 @@ The worker also requires: | --- | --- | | `BACKTEST_DATABASE_URL` | Shared PostgreSQL URL. | | `BACKTEST_WORKER_ID` | Stable instance identity for attempt evidence. | +| `BACKTEST_LOG_LEVEL` | Optional Python logging level name; defaults to `INFO`. Invalid names fail startup. Retry transitions include run, attempt, receive count, message, and reason identifiers without payloads or credentials. | | `BACKTEST_RESULTS_BUCKET`, `BACKTEST_RESULTS_PREFIX` | Same result store as the API. | | `BACKTEST_MARKET_DATA_BUCKET` | Immutable market-data input bucket. | | `BACKTEST_MARKET_DATA_CACHE` | Writable private cache directory, normally `/tmp/idea2strategy-market-data`. | @@ -123,6 +124,11 @@ The worker also requires: | `BACKTEST_RUNTIME_POLICY_FILE` | Read-only versioned attempt, microstructure, fractional eligibility, and risk-limit document. | | `BACKTEST_WORKER_CORRELATION_ID` | Deployment-operation correlation identifier. | +Every retry closes the current `backtest.run_attempts` row with +`terminal_reason_code=RETRY_RELEASED` and preserves the handler or orchestrator reason in +`failure_code`; the owning `backtest.runs` row remains non-terminal so a later delivery may create +the next fenced attempt. + AWS credentials are resolved by the SDK credential chain. On EC2, use the instance profile; do not inject access keys. `AWS_REGION` selects the region and `AWS_ENDPOINT_URL` is only for a local S3/SQS emulator. When local S3 and SQS diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index f28f3b8..00ec5c7 100644 --- a/src/backtest_engine/wiring.py +++ b/src/backtest_engine/wiring.py @@ -782,10 +782,22 @@ def heartbeat(self, key: str, claim: ExecutionClaim, *, lease_duration: timedelt with self._persistence.unit_of_work() as uow: uow.attempts.heartbeat_fenced(attempt_id, claim_token, lease_duration=lease_duration) - def release(self, key: str, *, now: datetime, claim: ExecutionClaim | None = None) -> None: + def release( + self, + key: str, + *, + now: datetime, + claim: ExecutionClaim | None = None, + reason_code: str | None = None, + ) -> None: attempt_id, claim_token = self._claim_ids(claim) with self._persistence.unit_of_work() as uow: - uow.attempts.release_fenced(attempt_id, claim_token, terminal_reason_code="RETRY_RELEASED") + uow.attempts.release_fenced( + attempt_id, + claim_token, + terminal_reason_code="RETRY_RELEASED", + failure_code=reason_code or "RETRY_REQUESTED", + ) def finish( self, diff --git a/src/backtest_engine/worker.py b/src/backtest_engine/worker.py index c535fd5..f98d3b7 100644 --- a/src/backtest_engine/worker.py +++ b/src/backtest_engine/worker.py @@ -174,7 +174,14 @@ def claim( def heartbeat(self, key: str, claim: ExecutionClaim, *, lease_duration: timedelta) -> None: ... - def release(self, key: str, *, now: datetime, claim: ExecutionClaim | None = None) -> None: ... + def release( + self, + key: str, + *, + now: datetime, + claim: ExecutionClaim | None = None, + reason_code: str | None = None, + ) -> None: ... def finish( self, @@ -236,7 +243,14 @@ def claim( def heartbeat(self, key: str, claim: ExecutionClaim, *, lease_duration: timedelta) -> None: return None - def release(self, key: str, *, now: datetime, claim: ExecutionClaim | None = None) -> None: + def release( + self, + key: str, + *, + now: datetime, + claim: ExecutionClaim | None = None, + reason_code: str | None = None, + ) -> None: """Hand a retryable attempt back so the next delivery can re-claim it.""" with self._lock: record = self._records.pop(key, None) @@ -487,9 +501,24 @@ def _handle(self, message: Mapping[str, Any]) -> HandledMessage: if outcome.result is JobResult.RETRY: # Release the CAS record so the redelivery is a *new* attempt, then # make the message immediately visible again. - self._store.release(key, now=self._clock(), claim=claim) + reason_code = outcome.reason_code or "RETRY_REQUESTED" + _LOGGER.info( + "backtest job retry released run_id=%s attempt=%s receive_count=%s " + "reason_code=%s message_id=%s", + run_id, + context.attempt_number, + context.receive_count, + reason_code, + message_id, + ) + self._store.release( + key, + now=self._clock(), + claim=claim, + reason_code=reason_code, + ) self._return_to_queue(receipt) - return HandledMessage(message_id, MessageDisposition.RETURNED, outcome.reason_code, key) + return HandledMessage(message_id, MessageDisposition.RETURNED, reason_code, key) self._store.finish(key, ExecutionRecordStatus.FAILED, now=self._clock(), claim=claim) return self._dead_letter(message, receipt, outcome.reason_code or "PERMANENT_FAILURE", message_id, key) @@ -509,6 +538,13 @@ def _invoke( try: return self._handler(job, context) except Exception as exc: + _LOGGER.exception( + "backtest job handler raised run_id=%s attempt=%s receive_count=%s message_id=%s", + job.get("backtestRunId"), + context.attempt_number, + context.receive_count, + context.message_id, + ) return JobOutcome(JobResult.RETRY, reason_code=f"HANDLER_ERROR:{type(exc).__name__}") finally: done.set() @@ -817,6 +853,20 @@ def _next_wait_seconds(self) -> float: BacktestLane.COMPETITION: 1, } +_LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s %(message)s" + + +def _configure_logging(environ: Mapping[str, str]) -> str: + """Configure the dedicated console process before any worker components start.""" + level_name = environ.get("BACKTEST_LOG_LEVEL", "INFO").strip().upper() or "INFO" + level = logging.getLevelNamesMapping().get(level_name) + if not isinstance(level, int): + raise WorkerConfigurationError( + f"BACKTEST_LOG_LEVEL must be a Python logging level name, got {level_name!r}" + ) + logging.basicConfig(level=level, format=_LOG_FORMAT, force=True) + return level_name + def _config_from_env(environ: Mapping[str, str]) -> WorkerConfig: missing = [name for name in _REQUIRED_ENV if not environ.get(name)] @@ -975,11 +1025,18 @@ def _runtime_sqs_client(environ: Mapping[str, str]) -> Any: def run() -> None: + log_level = _configure_logging(os.environ) lane_mode = any(os.environ.get(f"BACKTEST_{lane.value.upper()}_QUEUE_URL") for lane in BacktestLane) if lane_mode: configs, lane_limits, global_limit = _lane_configs_from_env(os.environ) else: config = _config_from_env(os.environ) + _LOGGER.info( + "backtest worker starting worker_id=%s lane_mode=%s log_level=%s", + os.environ.get("BACKTEST_WORKER_ID", ""), + lane_mode, + log_level, + ) handler: JobHandler = load_factory(os.environ["BACKTEST_JOB_HANDLER"], "BACKTEST_JOB_HANDLER") # No in-memory fallback. `InMemoryExecutionKeyStore` is a process-local dictionary; # a deployment that got it by leaving one variable unset would silently lose the diff --git a/tests/test_lane_scheduler.py b/tests/test_lane_scheduler.py index 5d65e61..9f77042 100644 --- a/tests/test_lane_scheduler.py +++ b/tests/test_lane_scheduler.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import logging import threading import time from collections import deque @@ -421,10 +422,28 @@ def test_scheduler_acknowledges_only_after_durable_finish() -> None: assert client.deleted == [message["ReceiptHandle"]] -def test_scheduler_retry_releases_the_claim_and_returns_message_to_its_lane() -> None: +def test_scheduler_retry_releases_the_claim_and_returns_message_to_its_lane( + caplog: pytest.LogCaptureFixture, +) -> None: message = _message(BacktestLane.CUSTOM, 2) client = OneMessageSqs(message) - store = InMemoryExecutionKeyStore() + + class RetryReasonStore(InMemoryExecutionKeyStore): + released_reason_code: str | None = None + + def release( + self, + key: str, + *, + now: datetime, + claim: Any = None, + reason_code: str | None = None, + ) -> None: + self.released_reason_code = reason_code + super().release(key, now=now, claim=claim, reason_code=reason_code) + + store = RetryReasonStore() + caplog.set_level(logging.INFO, logger="backtest_engine.worker") worker = BacktestWorker( client=client, config=WorkerConfig( @@ -453,3 +472,41 @@ def test_scheduler_retry_releases_the_claim_and_returns_message_to_its_lane() -> assert [item.disposition for item in completed] == [MessageDisposition.RETURNED] assert client.visibility_changes == [0] assert client.deleted == [] + assert store.released_reason_code == "TRANSIENT_DEPENDENCY" + assert "reason_code=TRANSIENT_DEPENDENCY" in caplog.text + + +def test_handler_exception_emits_traceback_and_a_stable_retry_reason( + caplog: pytest.LogCaptureFixture, +) -> None: + message = _message(BacktestLane.BASIC, 3) + client = OneMessageSqs(message) + + def fail(_job: Mapping[str, Any], _context: Any) -> JobOutcome: + raise OSError("simulated data read failure") + + worker = BacktestWorker( + client=client, + config=WorkerConfig( + queue_url="https://sqs.local/basic", + dead_letter_queue_url="https://sqs.local/basic-dlq", + worker_id="lane-worker", + max_receive_count=3, + visibility_timeout=timedelta(seconds=30), + wait_time=timedelta(0), + max_messages=1, + heartbeat_interval=timedelta(seconds=10), + ), + handler=fail, + store=InMemoryExecutionKeyStore(), + clock=lambda: T0, + ) + caplog.set_level(logging.INFO, logger="backtest_engine.worker") + + handled = worker.handle_message(message) + + assert handled.disposition is MessageDisposition.RETURNED + assert handled.reason_code == "HANDLER_ERROR:OSError" + assert "backtest job handler raised" in caplog.text + assert "simulated data read failure" in caplog.text + assert "reason_code=HANDLER_ERROR:OSError" in caplog.text diff --git a/tests/test_wiring_persistence.py b/tests/test_wiring_persistence.py index 1cb9629..30e6067 100644 --- a/tests/test_wiring_persistence.py +++ b/tests/test_wiring_persistence.py @@ -138,9 +138,26 @@ def test_a_released_attempt_is_closed_and_retry_gets_a_fresh_fence( key = worker_execution_key_for(str(run_id), "OFFICIAL_BACKTEST:retried") first = store.claim(key, run_id=str(run_id), owner="worker-a", now=T0, lease_duration=LEASE) - store.release(key, now=T0 + timedelta(seconds=5), claim=first) + store.release( + key, + now=T0 + timedelta(seconds=5), + claim=first, + reason_code="REQUIRED_DATA_UNAVAILABLE", + ) assert attempt_rows(admin_engine, run_id)[0]["status"] == "FAILED" assert store.status(key) is ExecutionRecordStatus.FAILED + with admin_engine.connect() as connection: + retry_reason = connection.execute( + text( + "SELECT failure_code, terminal_reason_code " + "FROM backtest.run_attempts WHERE id = CAST(:id AS uuid)" + ), + {"id": first.attempt_id}, + ).mappings().one() + assert dict(retry_reason) == { + "failure_code": "REQUIRED_DATA_UNAVAILABLE", + "terminal_reason_code": "RETRY_RELEASED", + } reclaimed = store.claim( key, run_id=str(run_id), owner="worker-b", now=T0 + timedelta(seconds=6), lease_duration=LEASE diff --git a/tests/test_worker.py b/tests/test_worker.py index 0a5be3a..12144fd 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -15,6 +15,7 @@ import contextlib import json +import logging import os import threading import time @@ -27,6 +28,7 @@ import boto3 import pytest +from backtest_engine import worker as worker_module from backtest_engine.worker import ( WORKER_EXECUTION_KEY_MAX_LENGTH, BacktestWorker, @@ -48,6 +50,38 @@ T0 = datetime(2026, 1, 1, tzinfo=timezone.utc) +@pytest.mark.parametrize( + ("environ", "expected_level"), + [ + ({}, logging.INFO), + ({"BACKTEST_LOG_LEVEL": "debug"}, logging.DEBUG), + ], +) +def test_worker_entrypoint_configures_runtime_logging( + monkeypatch: pytest.MonkeyPatch, + environ: Mapping[str, str], + expected_level: int, +) -> None: + configured: list[dict[str, object]] = [] + monkeypatch.setattr(worker_module.logging, "basicConfig", lambda **values: configured.append(values)) + + level_name = worker_module._configure_logging(environ) + + assert level_name == logging.getLevelName(expected_level) + assert configured == [ + { + "level": expected_level, + "format": "%(asctime)s %(levelname)s %(name)s %(message)s", + "force": True, + } + ] + + +def test_worker_entrypoint_rejects_an_unknown_log_level() -> None: + with pytest.raises(WorkerConfigurationError, match="BACKTEST_LOG_LEVEL"): + worker_module._configure_logging({"BACKTEST_LOG_LEVEL": "chatty"}) + + def test_sqs_client_uses_the_explicit_runtime_region(monkeypatch: pytest.MonkeyPatch) -> None: captured: dict[str, object] = {}