From 5aecac4464850b821c4bb87ef02bb3e26e673158 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 21 Jul 2026 04:49:47 +0800 Subject: [PATCH 1/6] fix: harden IBKR precheck runtime dispatch Co-Authored-By: Codex --- .github/workflows/sync-cloud-run-env.yml | 8 +- application/monitor_dispatcher.py | 124 +++++++++++++++++++--- application/runtime_deadline.py | 48 +++++++++ main.py | 63 +++++++++-- scripts/reconcile_cloud_runtime.py | 1 + tests/test_monitor_dispatcher.py | 72 ++++++++++++- tests/test_reconcile_cloud_runtime.py | 2 + tests/test_request_handling.py | 66 +++++++++++- tests/test_runtime_deadline.py | 22 ++++ tests/test_sync_cloud_run_env_workflow.sh | 2 + 10 files changed, 382 insertions(+), 26 deletions(-) create mode 100644 application/runtime_deadline.py create mode 100644 tests/test_runtime_deadline.py diff --git a/.github/workflows/sync-cloud-run-env.yml b/.github/workflows/sync-cloud-run-env.yml index bce082b..76bb607 100644 --- a/.github/workflows/sync-cloud-run-env.yml +++ b/.github/workflows/sync-cloud-run-env.yml @@ -907,13 +907,19 @@ jobs: plan = json.loads(os.environ["SYNC_PLAN_JSON"]) target = plan["targets"][int(os.environ["SYNC_TARGET_INDEX"])] env = target.get("env") or {} + scheduler = target.get("scheduler") or {} payload = { "service_name": target["service_name"], "service_url": os.environ["SERVICE_URL"].rstrip("/"), "strategy_profile": target.get("strategy_profile") or env.get("STRATEGY_PROFILE"), "account_group": env.get("ACCOUNT_GROUP"), "runtime_target_enabled": env.get("RUNTIME_TARGET_ENABLED", "true"), - "scheduler": target.get("scheduler") or {}, + # Dedicated Cloud Scheduler jobs already own /health warmups and + # /run execution. The shared dispatcher only owns prechecks. + "scheduler": { + "timezone": scheduler.get("timezone"), + "precheck_time": scheduler.get("precheck_time"), + }, } print(json.dumps(payload, separators=(",", ":"), sort_keys=True)) PY diff --git a/application/monitor_dispatcher.py b/application/monitor_dispatcher.py index 3e5dbc8..03af8c1 100644 --- a/application/monitor_dispatcher.py +++ b/application/monitor_dispatcher.py @@ -3,11 +3,15 @@ import json import os +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime +from datetime import timezone from typing import Any from quant_platform_kit.common.platform_runner.monitor import ( + _as_utc, + _iter_due_dispatches, dispatch_due_monitors as _dispatch_due_monitors, - load_monitor_targets as _load_targets_from_env, ) @@ -27,26 +31,118 @@ def max_workers_from_env() -> int: def load_monitor_targets(raw_json: str | None = None) -> list[dict[str, Any]]: """Read target config from env or from a direct JSON string.""" - if raw_json: - return json.loads(raw_json) - env = os.environ if "IBKR_MONITOR_DISPATCH_TARGETS_JSON" in os.environ else None - return _load_targets_from_env(env=env) + raw = str(raw_json or os.environ.get("IBKR_MONITOR_DISPATCH_TARGETS_JSON") or "").strip() + if not raw: + return [] + payload = json.loads(raw) + targets = payload.get("targets") if isinstance(payload, dict) else payload + if not isinstance(targets, list): + raise ValueError("IBKR monitor dispatch targets must be a JSON array or an object with targets") + return [dict(target) for target in targets if isinstance(target, dict)] + + +def due_monitor_dispatches( + targets: list[dict[str, Any]], + *, + now: datetime, + lookback_minutes: int, +) -> list[dict[str, Any]]: + """Expose shared due-window selection through the IBKR adapter.""" + return list( + _iter_due_dispatches( + targets, + now_utc=_as_utc(now), + lookback_minutes=lookback_minutes, + ) + ) def dispatch_due_monitor_targets( targets: list[dict[str, Any]], *, - now: str | None = None, + now: str | datetime | None = None, lookback_minutes: int | None = None, timeout_seconds: int | None = None, max_workers: int | None = None, + token_fetcher=None, + request_fn=None, + local_service_name: str | None = None, + local_dispatch_fn=None, ) -> dict[str, Any]: - from datetime import datetime - - return _dispatch_due_monitors( - targets, - now=datetime.fromisoformat(now) if now else None, - lookback_minutes=lookback_minutes or lookback_minutes_from_env(), - timeout_seconds=timeout_seconds or timeout_seconds_from_env(), - max_workers=max_workers or max_workers_from_env(), + resolved_now = datetime.fromisoformat(now) if isinstance(now, str) else now + resolved_now = resolved_now or datetime.now(timezone.utc) + resolved_lookback = lookback_minutes or lookback_minutes_from_env() + local_name = str(local_service_name or "").strip() + if not local_name or local_dispatch_fn is None: + result = _dispatch_due_monitors( + targets, + now=resolved_now, + lookback_minutes=resolved_lookback, + timeout_seconds=timeout_seconds or timeout_seconds_from_env(), + max_workers=max_workers or max_workers_from_env(), + token_fetcher=token_fetcher, + post_fn=request_fn, + ) + results = list(result.get("results") or []) + return {**result, "dispatches_sent": len(results)} + + local_targets = [target for target in targets if str(target.get("service_name") or "").strip() == local_name] + remote_targets = [target for target in targets if str(target.get("service_name") or "").strip() != local_name] + local_due = due_monitor_dispatches( + local_targets, + now=resolved_now, + lookback_minutes=resolved_lookback, ) + + # Run remote HTTP dispatches concurrently while the host service executes its + # own due check in-process. This avoids a self-call deadlock on maxScale=1, + # single-worker Cloud Run services without serially adding both timeout budgets. + with ThreadPoolExecutor(max_workers=1) as executor: + remote_future = executor.submit( + _dispatch_due_monitors, + remote_targets, + now=resolved_now, + lookback_minutes=resolved_lookback, + timeout_seconds=timeout_seconds or timeout_seconds_from_env(), + max_workers=max_workers or max_workers_from_env(), + token_fetcher=token_fetcher, + post_fn=request_fn, + ) + local_results = [_dispatch_local(dispatch, local_dispatch_fn) for dispatch in local_due] + remote_result = remote_future.result() + + results = [*local_results, *(remote_result.get("results") or [])] + results.sort(key=lambda item: (str(item.get("service_name") or ""), str(item.get("window") or ""))) + return { + "ok": all(bool(result.get("ok")) for result in results), + "total_targets": len(targets), + "dispatches_due": len(local_due) + int(remote_result.get("dispatches_due") or 0), + "dispatches_sent": len(results), + "results": results, + } + + +def _dispatch_local(dispatch: dict[str, Any], local_dispatch_fn) -> dict[str, Any]: + base_result = { + key: dispatch.get(key) + for key in ("service_name", "strategy_profile", "account_scope", "window", "url") + } + try: + outcome = dict(local_dispatch_fn(dispatch) or {}) + status_code = int(outcome.get("status_code") or 0) + return { + **base_result, + **outcome, + "status_code": status_code, + "ok": 200 <= status_code < 300, + "dispatch_mode": "in_process", + } + except Exception as exc: + return { + **base_result, + "status_code": 0, + "ok": False, + "dispatch_mode": "in_process", + "error_type": type(exc).__name__, + "error": str(exc), + } diff --git a/application/runtime_deadline.py b/application/runtime_deadline.py new file mode 100644 index 0000000..dfc1d97 --- /dev/null +++ b/application/runtime_deadline.py @@ -0,0 +1,48 @@ +"""Bound synchronous Cloud Run work before the platform request deadline.""" + +from __future__ import annotations + +import contextlib +import signal +import threading +from collections.abc import Iterator + + +class RuntimeDeadlineExceeded(TimeoutError): + """Raised when a synchronous runtime operation exceeds its local budget.""" + + +@contextlib.contextmanager +def enforce_runtime_deadline(seconds: int, *, operation: str) -> Iterator[None]: + """Interrupt main-thread work on POSIX before Cloud Run kills the worker.""" + timeout_seconds = int(seconds) + can_arm = ( + timeout_seconds > 0 + and threading.current_thread() is threading.main_thread() + and hasattr(signal, "SIGALRM") + and hasattr(signal, "setitimer") + ) + if not can_arm: + yield + return + + previous_delay, _previous_interval = signal.getitimer(signal.ITIMER_REAL) + if previous_delay > 0: + # Do not replace a deadline owned by the process manager or caller. + yield + return + + previous_handler = signal.getsignal(signal.SIGALRM) + + def _raise_deadline(_signum, _frame) -> None: + raise RuntimeDeadlineExceeded( + f"{operation} exceeded the {timeout_seconds}s application deadline" + ) + + signal.signal(signal.SIGALRM, _raise_deadline) + signal.setitimer(signal.ITIMER_REAL, timeout_seconds) + try: + yield + finally: + signal.setitimer(signal.ITIMER_REAL, 0) + signal.signal(signal.SIGALRM, previous_handler) diff --git a/main.py b/main.py index 8be540d..134de73 100644 --- a/main.py +++ b/main.py @@ -21,6 +21,7 @@ from application.cycle_result import coerce_strategy_cycle_result from application.runtime_broker_adapters import build_runtime_broker_adapters from application.runtime_composer import build_runtime_composer +from application.runtime_deadline import RuntimeDeadlineExceeded, enforce_runtime_deadline from application.runtime_strategy_adapters import ( build_runtime_strategy_adapters, fetch_yfinance_historical_candles, @@ -206,6 +207,7 @@ def _split_env_list(value: str | None) -> tuple[str, ...]: IB_CONNECT_ATTEMPTS = get_positive_int_env("IBKR_CONNECT_ATTEMPTS", 3) IB_CONNECT_RETRY_DELAY_SECONDS = get_non_negative_float_env("IBKR_CONNECT_RETRY_DELAY_SECONDS", 5.0) IB_CLIENT_ID_RETRY_OFFSET = get_positive_int_env("IBKR_CLIENT_ID_RETRY_OFFSET", 100) +IBKR_DRY_RUN_DEADLINE_SECONDS = get_positive_int_env("IBKR_DRY_RUN_DEADLINE_SECONDS", 105) STRATEGY_PROFILE = RUNTIME_SETTINGS.strategy_profile STRATEGY_DISPLAY_NAME = RUNTIME_SETTINGS.strategy_display_name @@ -1198,14 +1200,18 @@ def _handle_request( if dry_run_only_override is None: publish_strategy_plugin_alerts(strategy_plugin_signals, report=report) notification_delivery_events: list[dict] = [] - cycle_result = coerce_strategy_cycle_result( - run_strategy_core( - strategy_plugin_signals=strategy_plugin_signals, - strategy_plugin_error=strategy_plugin_error, - dry_run_only_override=dry_run_only_override, - notification_delivery_events=notification_delivery_events, + with enforce_runtime_deadline( + IBKR_DRY_RUN_DEADLINE_SECONDS if dry_run_only_override is True else 0, + operation="IBKR strategy dry-run", + ): + cycle_result = coerce_strategy_cycle_result( + run_strategy_core( + strategy_plugin_signals=strategy_plugin_signals, + strategy_plugin_error=strategy_plugin_error, + dry_run_only_override=dry_run_only_override, + notification_delivery_events=notification_delivery_events, + ) ) - ) execution_summary = dict(cycle_result.execution_summary or {}) reconciliation_record = dict(cycle_result.reconciliation_record or {}) signal_metadata = dict(cycle_result.signal_metadata or {}) @@ -1277,6 +1283,30 @@ def _handle_request( result=cycle_result.result, ) return (response_body if dry_run_only_override else cycle_result.result), 200 + except RuntimeDeadlineExceeded as exc: + append_runtime_report_error( + report, + stage="strategy_deadline", + message=str(exc), + error_type=type(exc).__name__, + ) + finalize_runtime_report(report, status="error") + log_runtime_event( + log_context, + "strategy_cycle_deadline_exceeded", + message="Strategy dry-run exceeded its application deadline", + severity="ERROR", + execution_window=execution_window, + timeout_seconds=IBKR_DRY_RUN_DEADLINE_SECONDS, + error_message=str(exc), + ) + error_msg = f"🚨 【IBKR 运行超时】\n{str(exc)}" + _publish_runtime_failure_notification( + detailed_text=error_msg, + compact_text=error_msg, + exc=exc, + ) + return "Error", 503 except TimeoutError as exc: append_runtime_report_error( report, @@ -1457,6 +1487,8 @@ def _handle_monitor_dispatch(): lookback_minutes=lookback_minutes_from_env(), timeout_seconds=timeout_seconds_from_env(), max_workers=max_workers_from_env(), + local_service_name=SERVICE_NAME or os.getenv("K_SERVICE"), + local_dispatch_fn=_dispatch_local_monitor, ) log_runtime_event( log_context, @@ -1467,7 +1499,22 @@ def _handle_monitor_dispatch(): dispatches_sent=result.get("dispatches_sent"), dispatch_results=result.get("results") or [], ) - return result, 200 + return result, 200 if result.get("ok") else 502 + + +def _dispatch_local_monitor(dispatch): + window = str(dispatch.get("window") or "").strip() + if window == "probe": + _body, status_code = _handle_probe() + elif window == "precheck": + _body, status_code = _handle_request( + dry_run_only_override=True, + response_body="Dry Run OK", + dry_run_label="strategy dry-run", + ) + else: + raise ValueError(f"Unsupported local monitor window: {window!r}") + return {"status_code": int(status_code)} @app.route("/run", methods=["POST", "GET"]) diff --git a/scripts/reconcile_cloud_runtime.py b/scripts/reconcile_cloud_runtime.py index 973c2d8..219ae53 100755 --- a/scripts/reconcile_cloud_runtime.py +++ b/scripts/reconcile_cloud_runtime.py @@ -299,6 +299,7 @@ def _legacy_jobs_for_target(platform: str, target: RuntimeTarget) -> list[str]: suffix = suffix[: -len("-service")] if suffix.startswith("u"): jobs.append(f"ibkr-{suffix}-backup-execution") + jobs.append(f"ibkr-{suffix}-pre-market-dry-run") return list(dict.fromkeys(jobs)) diff --git a/tests/test_monitor_dispatcher.py b/tests/test_monitor_dispatcher.py index 0822ff7..415a629 100644 --- a/tests/test_monitor_dispatcher.py +++ b/tests/test_monitor_dispatcher.py @@ -1,6 +1,19 @@ import datetime as dt -from application.monitor_dispatcher import dispatch_due_monitor_targets +from application.monitor_dispatcher import ( + dispatch_due_monitor_targets, + due_monitor_dispatches, + load_monitor_targets, +) + + +def test_load_monitor_targets_reads_ibkr_specific_env(monkeypatch): + monkeypatch.setenv( + "IBKR_MONITOR_DISPATCH_TARGETS_JSON", + '{"targets":[{"service_name":"svc-tqqq"}]}', + ) + + assert load_monitor_targets() == [{"service_name": "svc-tqqq"}] def test_due_monitor_dispatches_selects_due_window_and_skips_disabled_target(): @@ -75,8 +88,63 @@ def fake_request(url, *, headers, timeout): "https://svc-tqqq.example.run.app/probe", { "Authorization": "Bearer token-for:https://svc-tqqq.example.run.app", - "User-Agent": "ibkr-monitor-dispatcher", + "User-Agent": "platform-monitor-dispatcher", }, 12, ) ] + + +def test_dispatch_failure_is_exposed_in_result(): + def failing_request(_url, *, headers, timeout): + raise TimeoutError(f"request exceeded {timeout}s") + + result = dispatch_due_monitor_targets( + [ + { + "service_name": "svc-tqqq", + "service_url": "https://svc-tqqq.example.run.app", + "scheduler": { + "timezone": "America/New_York", + "precheck_time": "45 9 * * *", + }, + } + ], + now="2026-06-18T13:45:00+00:00", + request_fn=failing_request, + token_fetcher=lambda _audience: "token", + timeout_seconds=12, + ) + + assert result["ok"] is False + assert result["dispatches_sent"] == 1 + assert result["results"][0]["error_type"] == "TimeoutError" + + +def test_local_target_dispatches_in_process_without_self_http_call(): + http_calls = [] + local_calls = [] + + result = dispatch_due_monitor_targets( + [ + { + "service_name": "host-service", + "service_url": "https://host-service.example.run.app", + "scheduler": { + "timezone": "America/New_York", + "precheck_time": "45 9 * * *", + }, + } + ], + now="2026-06-18T13:45:00+00:00", + request_fn=lambda *args, **kwargs: http_calls.append((args, kwargs)), + local_service_name="host-service", + local_dispatch_fn=lambda dispatch: local_calls.append(dispatch) or {"status_code": 200}, + ) + + assert result["ok"] is True + assert result["dispatches_due"] == 1 + assert result["dispatches_sent"] == 1 + assert result["results"][0]["dispatch_mode"] == "in_process" + assert local_calls[0]["window"] == "precheck" + assert http_calls == [] diff --git a/tests/test_reconcile_cloud_runtime.py b/tests/test_reconcile_cloud_runtime.py index 8a92475..4e990e1 100644 --- a/tests/test_reconcile_cloud_runtime.py +++ b/tests/test_reconcile_cloud_runtime.py @@ -62,6 +62,7 @@ def test_legacy_jobs_for_ibkr_service_include_only_explicit_candidates(self) -> "interactive-brokers-quant-live-u1234-probe-scheduler", "interactive-brokers-quant-live-u1234-precheck-scheduler", "ibkr-u1234-backup-execution", + "ibkr-u1234-pre-market-dry-run", }, ) @@ -182,6 +183,7 @@ def test_delete_legacy_schedulers_deletes_only_known_jobs(self) -> None: "interactive-brokers-quant-live-u1234-probe-scheduler", "interactive-brokers-quant-live-u1234-precheck-scheduler", "ibkr-u1234-backup-execution", + "ibkr-u1234-pre-market-dry-run", } describe_calls: list[list[str]] = [] delete_calls: list[list[str]] = [] diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 91b0c97..f112e24 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -153,6 +153,48 @@ def fake_run_strategy_core(**kwargs): assert observed["events"][0][1]["execution_window"] == "dry_run" +def test_handle_dry_run_returns_service_unavailable_before_platform_timeout(strategy_module, monkeypatch): + observed = {"events": [], "notifications": []} + + class ExpiredDeadline: + def __enter__(self): + raise strategy_module.RuntimeDeadlineExceeded("dry-run deadline exceeded") + + def __exit__(self, *_args): + return False + + monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001")) + monkeypatch.setattr(strategy_module, "build_execution_report", lambda *_args, **_kwargs: {"status": "pending", "errors": []}) + monkeypatch.setattr( + strategy_module, + "persist_execution_report", + lambda report, **_kwargs: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json", + ) + monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True) + monkeypatch.setattr(strategy_module, "load_strategy_plugin_signals", lambda: ((), None)) + monkeypatch.setattr(strategy_module, "attach_strategy_plugin_report", lambda *_args, **_kwargs: None) + monkeypatch.setattr(strategy_module, "enforce_runtime_deadline", lambda *_args, **_kwargs: ExpiredDeadline()) + monkeypatch.setattr( + strategy_module, + "log_runtime_event", + lambda _context, event, **fields: observed["events"].append((event, fields)), + ) + monkeypatch.setattr( + strategy_module, + "_publish_runtime_failure_notification", + lambda **kwargs: observed["notifications"].append(kwargs) or True, + ) + + with strategy_module.app.test_request_context("/dry-run", method="POST"): + body, status = strategy_module.handle_dry_run() + + assert status == 503 + assert body == "Error" + assert observed["report"]["errors"][0]["stage"] == "strategy_deadline" + assert observed["events"][-1][0] == "strategy_cycle_deadline_exceeded" + assert len(observed["notifications"]) == 1 + + def test_dry_run_composer_wires_dry_run_override_to_order_execution(strategy_module, monkeypatch): observed = {} @@ -223,7 +265,7 @@ def test_handle_monitor_dispatch_post_dispatches_due_targets(strategy_module, mo def fake_dispatch(targets, **kwargs): observed["targets"] = targets observed["kwargs"] = kwargs - return {"dispatches_due": 1, "dispatches_sent": 1, "results": [{"service_name": "svc"}]} + return {"ok": True, "dispatches_due": 1, "dispatches_sent": 1, "results": [{"service_name": "svc"}]} monkeypatch.setattr(strategy_module, "dispatch_due_monitor_targets", fake_dispatch) monkeypatch.setattr( @@ -241,6 +283,28 @@ def fake_dispatch(targets, **kwargs): assert observed["events"][0][0] == "monitor_dispatch_completed" +def test_handle_monitor_dispatch_returns_bad_gateway_when_a_target_fails(strategy_module, monkeypatch): + monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001")) + monkeypatch.setattr(strategy_module, "load_monitor_targets", lambda: [{"service_name": "svc"}]) + monkeypatch.setattr( + strategy_module, + "dispatch_due_monitor_targets", + lambda *_args, **_kwargs: { + "ok": False, + "dispatches_due": 1, + "dispatches_sent": 1, + "results": [{"service_name": "svc", "status_code": 503, "ok": False}], + }, + ) + monkeypatch.setattr(strategy_module, "log_runtime_event", lambda *_args, **_kwargs: None) + + with strategy_module.app.test_request_context("/monitor-dispatch", method="POST"): + body, status = strategy_module.handle_monitor_dispatch() + + assert status == 502 + assert body["ok"] is False + + def test_handle_probe_checks_account_snapshot_without_success_notification(strategy_module, monkeypatch): observed = {"events": [], "disconnects": 0, "notifications": []} diff --git a/tests/test_runtime_deadline.py b/tests/test_runtime_deadline.py new file mode 100644 index 0000000..0bab57d --- /dev/null +++ b/tests/test_runtime_deadline.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +import time + +import pytest + +from application.runtime_deadline import RuntimeDeadlineExceeded, enforce_runtime_deadline + + +def test_runtime_deadline_interrupts_stalled_main_thread_work() -> None: + started = time.monotonic() + + with pytest.raises(RuntimeDeadlineExceeded, match="test operation exceeded the 1s"): + with enforce_runtime_deadline(1, operation="test operation"): + time.sleep(5) + + assert time.monotonic() - started < 2 + + +def test_disabled_runtime_deadline_does_not_interrupt() -> None: + with enforce_runtime_deadline(0, operation="disabled operation"): + pass diff --git a/tests/test_sync_cloud_run_env_workflow.sh b/tests/test_sync_cloud_run_env_workflow.sh index 03478cd..63f30d2 100644 --- a/tests/test_sync_cloud_run_env_workflow.sh +++ b/tests/test_sync_cloud_run_env_workflow.sh @@ -121,6 +121,8 @@ grep -Fq -- '--update-env-vars "^|^$(join_by_delimiter "|" "${env_pairs[@]}")' " grep -Fq -- '--update-labels "$(IFS=,; echo "${target_label_pairs[*]}")' "$workflow_file" grep -Fq 'monitor_dispatch_targets_json="$(jq -sc '\''{targets:.}'\'' "${monitor_dispatch_targets_file}")"' "$workflow_file" grep -Fq 'shared_env_pairs+=("IBKR_MONITOR_DISPATCH_TARGETS_JSON=${monitor_dispatch_targets_json}")' "$workflow_file" +grep -Fq '"precheck_time": scheduler.get("precheck_time")' "$workflow_file" +test "$(grep -Fc '"probe_time": scheduler.get("probe_time")' "$workflow_file")" -eq 0 grep -Fq 'Sync Cloud Scheduler schedule' "$workflow_file" grep -Fq 'scheduler_location="${CLOUD_SCHEDULER_LOCATION:-${CLOUD_RUN_REGION}}"' "$workflow_file" grep -Fq 'Cloud Scheduler schedule sync requires GCP_SCHEDULER_SERVICE_ACCOUNT.' "$workflow_file" From 8c848f37366ea7a63847474df16af3eb04fe8bd7 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:18:24 +0800 Subject: [PATCH 2/6] fix: recycle workers after dry-run deadline Co-Authored-By: Codex --- application/runtime_deadline.py | 33 ++++++++- main.py | 115 +++++++++++++++++--------------- tests/test_request_handling.py | 45 +++++++++---- tests/test_runtime_deadline.py | 19 +++++- 4 files changed, 141 insertions(+), 71 deletions(-) diff --git a/application/runtime_deadline.py b/application/runtime_deadline.py index dfc1d97..9636595 100644 --- a/application/runtime_deadline.py +++ b/application/runtime_deadline.py @@ -3,13 +3,20 @@ from __future__ import annotations import contextlib +import os import signal import threading +import time from collections.abc import Iterator +from collections.abc import Callable -class RuntimeDeadlineExceeded(TimeoutError): - """Raised when a synchronous runtime operation exceeds its local budget.""" +class RuntimeDeadlineExceeded(BaseException): + """Raised when a synchronous runtime operation exceeds its local budget. + + This intentionally bypasses broad ``except Exception`` handlers in broker and + strategy code so an interrupted worker cannot resume with partial session state. + """ @contextlib.contextmanager @@ -46,3 +53,25 @@ def _raise_deadline(_signum, _frame) -> None: finally: signal.setitimer(signal.ITIMER_REAL, 0) signal.signal(signal.SIGALRM, previous_handler) + + +def recycle_current_process_after_response( + *, + delay_seconds: float = 1.0, + process_id: int | None = None, + kill_fn: Callable[[int, int], None] = os.kill, +) -> threading.Thread: + """Ask the process manager to replace a worker after its response can flush.""" + target_pid = int(process_id or os.getpid()) + + def _terminate() -> None: + time.sleep(max(0.0, float(delay_seconds))) + kill_fn(target_pid, signal.SIGTERM) + + thread = threading.Thread( + target=_terminate, + name="runtime-deadline-worker-recycle", + daemon=True, + ) + thread.start() + return thread diff --git a/main.py b/main.py index 134de73..d6d808e 100644 --- a/main.py +++ b/main.py @@ -21,7 +21,11 @@ from application.cycle_result import coerce_strategy_cycle_result from application.runtime_broker_adapters import build_runtime_broker_adapters from application.runtime_composer import build_runtime_composer -from application.runtime_deadline import RuntimeDeadlineExceeded, enforce_runtime_deadline +from application.runtime_deadline import ( + RuntimeDeadlineExceeded, + enforce_runtime_deadline, + recycle_current_process_after_response, +) from application.runtime_strategy_adapters import ( build_runtime_strategy_adapters, fetch_yfinance_historical_candles, @@ -1113,6 +1117,7 @@ def _handle_request( ) execution_window = "dry_run" if dry_run_only_override else "execution" lock_acquired = STRATEGY_RUN_LOCK.acquire(blocking=False) + deadline_exceeded = False try: log_runtime_event( log_context, @@ -1200,18 +1205,14 @@ def _handle_request( if dry_run_only_override is None: publish_strategy_plugin_alerts(strategy_plugin_signals, report=report) notification_delivery_events: list[dict] = [] - with enforce_runtime_deadline( - IBKR_DRY_RUN_DEADLINE_SECONDS if dry_run_only_override is True else 0, - operation="IBKR strategy dry-run", - ): - cycle_result = coerce_strategy_cycle_result( - run_strategy_core( - strategy_plugin_signals=strategy_plugin_signals, - strategy_plugin_error=strategy_plugin_error, - dry_run_only_override=dry_run_only_override, - notification_delivery_events=notification_delivery_events, - ) + cycle_result = coerce_strategy_cycle_result( + run_strategy_core( + strategy_plugin_signals=strategy_plugin_signals, + strategy_plugin_error=strategy_plugin_error, + dry_run_only_override=dry_run_only_override, + notification_delivery_events=notification_delivery_events, ) + ) execution_summary = dict(cycle_result.execution_summary or {}) reconciliation_record = dict(cycle_result.reconciliation_record or {}) signal_metadata = dict(cycle_result.signal_metadata or {}) @@ -1283,30 +1284,9 @@ def _handle_request( result=cycle_result.result, ) return (response_body if dry_run_only_override else cycle_result.result), 200 - except RuntimeDeadlineExceeded as exc: - append_runtime_report_error( - report, - stage="strategy_deadline", - message=str(exc), - error_type=type(exc).__name__, - ) - finalize_runtime_report(report, status="error") - log_runtime_event( - log_context, - "strategy_cycle_deadline_exceeded", - message="Strategy dry-run exceeded its application deadline", - severity="ERROR", - execution_window=execution_window, - timeout_seconds=IBKR_DRY_RUN_DEADLINE_SECONDS, - error_message=str(exc), - ) - error_msg = f"🚨 【IBKR 运行超时】\n{str(exc)}" - _publish_runtime_failure_notification( - detailed_text=error_msg, - compact_text=error_msg, - exc=exc, - ) - return "Error", 503 + except RuntimeDeadlineExceeded: + deadline_exceeded = True + raise except TimeoutError as exc: append_runtime_report_error( report, @@ -1356,14 +1336,15 @@ def _handle_request( finally: if lock_acquired: STRATEGY_RUN_LOCK.release() - try: - if dry_run_only_override is None: - report_path = persist_execution_report(report) - else: - report_path = persist_execution_report(report, dry_run_only_override=dry_run_only_override) - print(f"execution_report {report_path}", flush=True) - except Exception as persist_exc: - print(f"failed to persist execution report: {persist_exc}", flush=True) + if not deadline_exceeded: + try: + if dry_run_only_override is None: + report_path = persist_execution_report(report) + else: + report_path = persist_execution_report(report, dry_run_only_override=dry_run_only_override) + print(f"execution_report {report_path}", flush=True) + except Exception as persist_exc: + print(f"failed to persist execution report: {persist_exc}", flush=True) def _handle_probe(*, response_body: str = "Probe OK"): @@ -1507,11 +1488,7 @@ def _dispatch_local_monitor(dispatch): if window == "probe": _body, status_code = _handle_probe() elif window == "precheck": - _body, status_code = _handle_request( - dry_run_only_override=True, - response_body="Dry Run OK", - dry_run_label="strategy dry-run", - ) + _body, status_code = _handle_dry_run_with_deadline() else: raise ValueError(f"Unsupported local monitor window: {window!r}") return {"status_code": int(status_code)} @@ -1522,14 +1499,42 @@ def handle_request(): return _route_with_runtime_error_fallback(_handle_request) +def _handle_dry_run_with_deadline(): + try: + with enforce_runtime_deadline( + IBKR_DRY_RUN_DEADLINE_SECONDS, + operation="IBKR strategy dry-run request", + ): + return _route_with_runtime_error_fallback( + _handle_request, + dry_run_only_override=True, + response_body="Dry Run OK", + dry_run_label="strategy dry-run", + ) + except RuntimeDeadlineExceeded as exc: + print( + json.dumps( + { + "severity": "ERROR", + "event": "strategy_cycle_deadline_exceeded", + "message": str(exc), + "service_name": SERVICE_NAME or os.getenv("K_SERVICE"), + "strategy_profile": STRATEGY_PROFILE, + "timeout_seconds": IBKR_DRY_RUN_DEADLINE_SECONDS, + "worker_recycle_scheduled": True, + }, + ensure_ascii=False, + separators=(",", ":"), + ), + flush=True, + ) + recycle_current_process_after_response() + return "Error", 503 + + @app.route("/dry-run", methods=["POST", "GET"]) def handle_dry_run(): - return _route_with_runtime_error_fallback( - _handle_request, - dry_run_only_override=True, - response_body="Dry Run OK", - dry_run_label="strategy dry-run", - ) + return _handle_dry_run_with_deadline() @app.route("/probe", methods=["POST", "GET"]) diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index f112e24..e7dd6fd 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -154,7 +154,7 @@ def fake_run_strategy_core(**kwargs): def test_handle_dry_run_returns_service_unavailable_before_platform_timeout(strategy_module, monkeypatch): - observed = {"events": [], "notifications": []} + observed = {"recycles": 0} class ExpiredDeadline: def __enter__(self): @@ -163,26 +163,46 @@ def __enter__(self): def __exit__(self, *_args): return False - monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001")) - monkeypatch.setattr(strategy_module, "build_execution_report", lambda *_args, **_kwargs: {"status": "pending", "errors": []}) + monkeypatch.setattr(strategy_module, "enforce_runtime_deadline", lambda *_args, **_kwargs: ExpiredDeadline()) monkeypatch.setattr( strategy_module, - "persist_execution_report", - lambda report, **_kwargs: observed.setdefault("report", dict(report)) or "/tmp/runtime-report.json", + "recycle_current_process_after_response", + lambda: observed.__setitem__("recycles", observed["recycles"] + 1), ) + + with strategy_module.app.test_request_context("/dry-run", method="POST"): + body, status = strategy_module.handle_dry_run() + + assert status == 503 + assert body == "Error" + assert observed["recycles"] == 1 + + +def test_handle_dry_run_deadline_from_strategy_skips_persistence_and_recycles(strategy_module, monkeypatch): + observed = {"persist_calls": 0, "recycles": 0} + + monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001")) + monkeypatch.setattr(strategy_module, "build_execution_report", lambda *_args, **_kwargs: {"status": "pending"}) monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True) monkeypatch.setattr(strategy_module, "load_strategy_plugin_signals", lambda: ((), None)) monkeypatch.setattr(strategy_module, "attach_strategy_plugin_report", lambda *_args, **_kwargs: None) - monkeypatch.setattr(strategy_module, "enforce_runtime_deadline", lambda *_args, **_kwargs: ExpiredDeadline()) + monkeypatch.setattr(strategy_module, "log_runtime_event", lambda *_args, **_kwargs: None) monkeypatch.setattr( strategy_module, - "log_runtime_event", - lambda _context, event, **fields: observed["events"].append((event, fields)), + "run_strategy_core", + lambda **_kwargs: (_ for _ in ()).throw( + strategy_module.RuntimeDeadlineExceeded("strategy deadline exceeded") + ), + ) + monkeypatch.setattr( + strategy_module, + "persist_execution_report", + lambda *_args, **_kwargs: observed.__setitem__("persist_calls", observed["persist_calls"] + 1), ) monkeypatch.setattr( strategy_module, - "_publish_runtime_failure_notification", - lambda **kwargs: observed["notifications"].append(kwargs) or True, + "recycle_current_process_after_response", + lambda: observed.__setitem__("recycles", observed["recycles"] + 1), ) with strategy_module.app.test_request_context("/dry-run", method="POST"): @@ -190,9 +210,8 @@ def __exit__(self, *_args): assert status == 503 assert body == "Error" - assert observed["report"]["errors"][0]["stage"] == "strategy_deadline" - assert observed["events"][-1][0] == "strategy_cycle_deadline_exceeded" - assert len(observed["notifications"]) == 1 + assert observed["persist_calls"] == 0 + assert observed["recycles"] == 1 def test_dry_run_composer_wires_dry_run_override_to_order_execution(strategy_module, monkeypatch): diff --git a/tests/test_runtime_deadline.py b/tests/test_runtime_deadline.py index 0bab57d..54ca5f9 100644 --- a/tests/test_runtime_deadline.py +++ b/tests/test_runtime_deadline.py @@ -4,7 +4,11 @@ import pytest -from application.runtime_deadline import RuntimeDeadlineExceeded, enforce_runtime_deadline +from application.runtime_deadline import ( + RuntimeDeadlineExceeded, + enforce_runtime_deadline, + recycle_current_process_after_response, +) def test_runtime_deadline_interrupts_stalled_main_thread_work() -> None: @@ -20,3 +24,16 @@ def test_runtime_deadline_interrupts_stalled_main_thread_work() -> None: def test_disabled_runtime_deadline_does_not_interrupt() -> None: with enforce_runtime_deadline(0, operation="disabled operation"): pass + + +def test_recycle_current_process_sends_sigterm_after_delay() -> None: + calls = [] + + thread = recycle_current_process_after_response( + delay_seconds=0, + process_id=1234, + kill_fn=lambda process_id, signum: calls.append((process_id, signum)), + ) + thread.join(timeout=1) + + assert calls == [(1234, 15)] From 8146bac0d9f1602bfc87cedecf72af995c11bfed Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:23:35 +0800 Subject: [PATCH 3/6] fix: defer nested precheck worker recycle Co-Authored-By: Codex --- main.py | 60 +++++++++++++++++++++++++++---- tests/test_request_handling.py | 65 ++++++++++++++++++++++++++++++++-- 2 files changed, 117 insertions(+), 8 deletions(-) diff --git a/main.py b/main.py index d6d808e..438669b 100644 --- a/main.py +++ b/main.py @@ -212,6 +212,7 @@ def _split_env_list(value: str | None) -> tuple[str, ...]: IB_CONNECT_RETRY_DELAY_SECONDS = get_non_negative_float_env("IBKR_CONNECT_RETRY_DELAY_SECONDS", 5.0) IB_CLIENT_ID_RETRY_OFFSET = get_positive_int_env("IBKR_CLIENT_ID_RETRY_OFFSET", 100) IBKR_DRY_RUN_DEADLINE_SECONDS = get_positive_int_env("IBKR_DRY_RUN_DEADLINE_SECONDS", 105) +IBKR_DEADLINE_REPORT_GRACE_SECONDS = get_positive_int_env("IBKR_DEADLINE_REPORT_GRACE_SECONDS", 5) STRATEGY_PROFILE = RUNTIME_SETTINGS.strategy_profile STRATEGY_DISPLAY_NAME = RUNTIME_SETTINGS.strategy_display_name @@ -1284,8 +1285,36 @@ def _handle_request( result=cycle_result.result, ) return (response_body if dry_run_only_override else cycle_result.result), 200 - except RuntimeDeadlineExceeded: + except RuntimeDeadlineExceeded as exc: deadline_exceeded = True + append_runtime_report_error( + report, + stage="strategy_deadline", + message=str(exc), + error_type=type(exc).__name__, + ) + finalize_runtime_report( + report, + status="error", + diagnostics={ + "failure_category": "strategy_deadline", + "timeout_seconds": IBKR_DRY_RUN_DEADLINE_SECONDS, + }, + ) + try: + with enforce_runtime_deadline( + IBKR_DEADLINE_REPORT_GRACE_SECONDS, + operation="IBKR deadline report persistence", + ): + report_path = persist_execution_report( + report, + dry_run_only_override=dry_run_only_override, + ) + print(f"execution_report {report_path}", flush=True) + except RuntimeDeadlineExceeded as persist_exc: + print(f"deadline report persistence timed out: {persist_exc}", flush=True) + except Exception as persist_exc: + print(f"failed to persist deadline report: {persist_exc}", flush=True) raise except TimeoutError as exc: append_runtime_report_error( @@ -1480,6 +1509,8 @@ def _handle_monitor_dispatch(): dispatches_sent=result.get("dispatches_sent"), dispatch_results=result.get("results") or [], ) + if any(bool(item.get("worker_recycle_required")) for item in (result.get("results") or [])): + recycle_current_process_after_response() return result, 200 if result.get("ok") else 502 @@ -1487,11 +1518,20 @@ def _dispatch_local_monitor(dispatch): window = str(dispatch.get("window") or "").strip() if window == "probe": _body, status_code = _handle_probe() + recycle_required = False elif window == "precheck": - _body, status_code = _handle_dry_run_with_deadline() + timeout_state: dict[str, bool] = {} + _body, status_code = _handle_dry_run_with_deadline( + recycle_on_timeout=False, + timeout_state=timeout_state, + ) + recycle_required = bool(timeout_state.get("worker_recycle_required")) else: raise ValueError(f"Unsupported local monitor window: {window!r}") - return {"status_code": int(status_code)} + return { + "status_code": int(status_code), + "worker_recycle_required": recycle_required, + } @app.route("/run", methods=["POST", "GET"]) @@ -1499,7 +1539,11 @@ def handle_request(): return _route_with_runtime_error_fallback(_handle_request) -def _handle_dry_run_with_deadline(): +def _handle_dry_run_with_deadline( + *, + recycle_on_timeout: bool = True, + timeout_state: dict[str, bool] | None = None, +): try: with enforce_runtime_deadline( IBKR_DRY_RUN_DEADLINE_SECONDS, @@ -1521,14 +1565,18 @@ def _handle_dry_run_with_deadline(): "service_name": SERVICE_NAME or os.getenv("K_SERVICE"), "strategy_profile": STRATEGY_PROFILE, "timeout_seconds": IBKR_DRY_RUN_DEADLINE_SECONDS, - "worker_recycle_scheduled": True, + "worker_recycle_required": True, + "worker_recycle_scheduled": recycle_on_timeout, }, ensure_ascii=False, separators=(",", ":"), ), flush=True, ) - recycle_current_process_after_response() + if timeout_state is not None: + timeout_state["worker_recycle_required"] = True + if recycle_on_timeout: + recycle_current_process_after_response() return "Error", 503 diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index e7dd6fd..58234c5 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -178,7 +178,7 @@ def __exit__(self, *_args): assert observed["recycles"] == 1 -def test_handle_dry_run_deadline_from_strategy_skips_persistence_and_recycles(strategy_module, monkeypatch): +def test_handle_dry_run_deadline_from_strategy_persists_once_and_recycles(strategy_module, monkeypatch): observed = {"persist_calls": 0, "recycles": 0} monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001")) @@ -210,7 +210,7 @@ def test_handle_dry_run_deadline_from_strategy_skips_persistence_and_recycles(st assert status == 503 assert body == "Error" - assert observed["persist_calls"] == 0 + assert observed["persist_calls"] == 1 assert observed["recycles"] == 1 @@ -324,6 +324,67 @@ def test_handle_monitor_dispatch_returns_bad_gateway_when_a_target_fails(strateg assert body["ok"] is False +def test_handle_monitor_dispatch_recycles_only_after_local_timeout_is_aggregated(strategy_module, monkeypatch): + observed = {"recycles": 0, "logged": False} + + monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001")) + monkeypatch.setattr(strategy_module, "load_monitor_targets", lambda: [{"service_name": "host-service"}]) + monkeypatch.setattr( + strategy_module, + "dispatch_due_monitor_targets", + lambda *_args, **_kwargs: { + "ok": False, + "dispatches_due": 1, + "dispatches_sent": 1, + "results": [ + { + "service_name": "host-service", + "status_code": 503, + "ok": False, + "worker_recycle_required": True, + } + ], + }, + ) + monkeypatch.setattr( + strategy_module, + "log_runtime_event", + lambda *_args, **_kwargs: observed.__setitem__("logged", True), + ) + monkeypatch.setattr( + strategy_module, + "recycle_current_process_after_response", + lambda: observed.__setitem__("recycles", observed["recycles"] + 1), + ) + + with strategy_module.app.test_request_context("/monitor-dispatch", method="POST"): + body, status = strategy_module.handle_monitor_dispatch() + + assert status == 502 + assert body["results"][0]["worker_recycle_required"] is True + assert observed == {"recycles": 1, "logged": True} + + +def test_local_precheck_defers_worker_recycle_to_monitor_dispatch(strategy_module, monkeypatch): + observed = {"calls": 0} + + def fake_dry_run(*, recycle_on_timeout, timeout_state): + observed["calls"] += 1 + assert recycle_on_timeout is False + timeout_state["worker_recycle_required"] = True + return "Error", 503 + + monkeypatch.setattr(strategy_module, "_handle_dry_run_with_deadline", fake_dry_run) + + result = strategy_module._dispatch_local_monitor({"window": "precheck"}) + + assert result == { + "status_code": 503, + "worker_recycle_required": True, + } + assert observed["calls"] == 1 + + def test_handle_probe_checks_account_snapshot_without_success_notification(strategy_module, monkeypatch): observed = {"events": [], "disconnects": 0, "notifications": []} From 155aef446d99ec08d1e36e8f673450ab40370f85 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:30:30 +0800 Subject: [PATCH 4/6] fix: separate strategy and report deadlines Co-Authored-By: Codex --- application/monitor_dispatcher.py | 5 ++- main.py | 53 +++++++++++++++++++++---------- tests/test_monitor_dispatcher.py | 7 ++++ tests/test_request_handling.py | 45 ++++++++++++++++++++++++++ 4 files changed, 93 insertions(+), 17 deletions(-) diff --git a/application/monitor_dispatcher.py b/application/monitor_dispatcher.py index 03af8c1..05fc12a 100644 --- a/application/monitor_dispatcher.py +++ b/application/monitor_dispatcher.py @@ -38,7 +38,10 @@ def load_monitor_targets(raw_json: str | None = None) -> list[dict[str, Any]]: targets = payload.get("targets") if isinstance(payload, dict) else payload if not isinstance(targets, list): raise ValueError("IBKR monitor dispatch targets must be a JSON array or an object with targets") - return [dict(target) for target in targets if isinstance(target, dict)] + for index, target in enumerate(targets): + if not isinstance(target, dict): + raise ValueError(f"IBKR monitor dispatch target at index {index} must be an object") + return [dict(target) for target in targets] def due_monitor_dispatches( diff --git a/main.py b/main.py index 438669b..3aa4cd0 100644 --- a/main.py +++ b/main.py @@ -1100,6 +1100,8 @@ def _handle_request( dry_run_only_override: bool | None = None, response_body: str = "OK", dry_run_label: str = "strategy dry-run", + persist_report: bool = True, + report_collector: list[dict] | None = None, ): if request.method == "GET": if dry_run_only_override is None: @@ -1301,20 +1303,6 @@ def _handle_request( "timeout_seconds": IBKR_DRY_RUN_DEADLINE_SECONDS, }, ) - try: - with enforce_runtime_deadline( - IBKR_DEADLINE_REPORT_GRACE_SECONDS, - operation="IBKR deadline report persistence", - ): - report_path = persist_execution_report( - report, - dry_run_only_override=dry_run_only_override, - ) - print(f"execution_report {report_path}", flush=True) - except RuntimeDeadlineExceeded as persist_exc: - print(f"deadline report persistence timed out: {persist_exc}", flush=True) - except Exception as persist_exc: - print(f"failed to persist deadline report: {persist_exc}", flush=True) raise except TimeoutError as exc: append_runtime_report_error( @@ -1365,7 +1353,9 @@ def _handle_request( finally: if lock_acquired: STRATEGY_RUN_LOCK.release() - if not deadline_exceeded: + if report_collector is not None: + report_collector.append(report) + if not deadline_exceeded and persist_report: try: if dry_run_only_override is None: report_path = persist_execution_report(report) @@ -1544,18 +1534,22 @@ def _handle_dry_run_with_deadline( recycle_on_timeout: bool = True, timeout_state: dict[str, bool] | None = None, ): + reports: list[dict] = [] try: with enforce_runtime_deadline( IBKR_DRY_RUN_DEADLINE_SECONDS, operation="IBKR strategy dry-run request", ): - return _route_with_runtime_error_fallback( + response = _route_with_runtime_error_fallback( _handle_request, dry_run_only_override=True, response_body="Dry Run OK", dry_run_label="strategy dry-run", + persist_report=False, + report_collector=reports, ) except RuntimeDeadlineExceeded as exc: + _persist_dry_run_report_with_grace(reports, timeout_context=True) print( json.dumps( { @@ -1578,6 +1572,33 @@ def _handle_dry_run_with_deadline( if recycle_on_timeout: recycle_current_process_after_response() return "Error", 503 + else: + _persist_dry_run_report_with_grace(reports, timeout_context=False) + return response + + +def _persist_dry_run_report_with_grace( + reports: list[dict], + *, + timeout_context: bool, +) -> None: + if not reports: + return + operation = "IBKR timeout report persistence" if timeout_context else "IBKR dry-run report persistence" + try: + with enforce_runtime_deadline( + IBKR_DEADLINE_REPORT_GRACE_SECONDS, + operation=operation, + ): + report_path = persist_execution_report( + reports[-1], + dry_run_only_override=True, + ) + print(f"execution_report {report_path}", flush=True) + except RuntimeDeadlineExceeded as persist_exc: + print(f"dry-run report persistence timed out: {persist_exc}", flush=True) + except Exception as persist_exc: + print(f"failed to persist dry-run report: {persist_exc}", flush=True) @app.route("/dry-run", methods=["POST", "GET"]) diff --git a/tests/test_monitor_dispatcher.py b/tests/test_monitor_dispatcher.py index 415a629..9359b7d 100644 --- a/tests/test_monitor_dispatcher.py +++ b/tests/test_monitor_dispatcher.py @@ -1,5 +1,7 @@ import datetime as dt +import pytest + from application.monitor_dispatcher import ( dispatch_due_monitor_targets, due_monitor_dispatches, @@ -16,6 +18,11 @@ def test_load_monitor_targets_reads_ibkr_specific_env(monkeypatch): assert load_monitor_targets() == [{"service_name": "svc-tqqq"}] +def test_load_monitor_targets_rejects_non_object_items(): + with pytest.raises(ValueError, match="target at index 1 must be an object"): + load_monitor_targets('{"targets":[{"service_name":"svc-tqqq"},"bad-target"]}') + + def test_due_monitor_dispatches_selects_due_window_and_skips_disabled_target(): targets = [ { diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index 58234c5..a995ad8 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -214,6 +214,51 @@ def test_handle_dry_run_deadline_from_strategy_persists_once_and_recycles(strate assert observed["recycles"] == 1 +def test_successful_dry_run_report_timeout_does_not_flip_strategy_result(strategy_module, monkeypatch): + observed = {"persist_calls": 0, "recycles": 0} + + class DeadlineContext: + def __init__(self, operation): + self.operation = operation + + def __enter__(self): + if "report persistence" in self.operation: + raise strategy_module.RuntimeDeadlineExceeded("report persistence deadline exceeded") + + def __exit__(self, *_args): + return False + + monkeypatch.setattr( + strategy_module, + "enforce_runtime_deadline", + lambda _seconds, *, operation: DeadlineContext(operation), + ) + monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001")) + monkeypatch.setattr(strategy_module, "build_execution_report", lambda *_args, **_kwargs: {"status": "pending"}) + monkeypatch.setattr(strategy_module, "is_market_open_now", lambda **_kwargs: True) + monkeypatch.setattr(strategy_module, "load_strategy_plugin_signals", lambda: ((), None)) + monkeypatch.setattr(strategy_module, "attach_strategy_plugin_report", lambda *_args, **_kwargs: None) + monkeypatch.setattr(strategy_module, "log_runtime_event", lambda *_args, **_kwargs: None) + monkeypatch.setattr(strategy_module, "run_strategy_core", lambda **_kwargs: "OK - dry-run") + monkeypatch.setattr( + strategy_module, + "persist_execution_report", + lambda *_args, **_kwargs: observed.__setitem__("persist_calls", observed["persist_calls"] + 1), + ) + monkeypatch.setattr( + strategy_module, + "recycle_current_process_after_response", + lambda: observed.__setitem__("recycles", observed["recycles"] + 1), + ) + + with strategy_module.app.test_request_context("/dry-run", method="POST"): + body, status = strategy_module.handle_dry_run() + + assert status == 200 + assert body == "Dry Run OK" + assert observed == {"persist_calls": 0, "recycles": 0} + + def test_dry_run_composer_wires_dry_run_override_to_order_execution(strategy_module, monkeypatch): observed = {} From f0dd4095debffcfb5ebeee44a6147977ec9fc6e8 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:37:08 +0800 Subject: [PATCH 5/6] fix: preserve generic monitor target fallback Co-Authored-By: Codex --- application/monitor_dispatcher.py | 7 ++++++- tests/test_monitor_dispatcher.py | 23 +++++++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/application/monitor_dispatcher.py b/application/monitor_dispatcher.py index 05fc12a..6a43e15 100644 --- a/application/monitor_dispatcher.py +++ b/application/monitor_dispatcher.py @@ -31,7 +31,12 @@ def max_workers_from_env() -> int: def load_monitor_targets(raw_json: str | None = None) -> list[dict[str, Any]]: """Read target config from env or from a direct JSON string.""" - raw = str(raw_json or os.environ.get("IBKR_MONITOR_DISPATCH_TARGETS_JSON") or "").strip() + raw = str( + raw_json + or os.environ.get("IBKR_MONITOR_DISPATCH_TARGETS_JSON") + or os.environ.get("MONITOR_DISPATCH_TARGETS_JSON") + or "" + ).strip() if not raw: return [] payload = json.loads(raw) diff --git a/tests/test_monitor_dispatcher.py b/tests/test_monitor_dispatcher.py index 9359b7d..9279056 100644 --- a/tests/test_monitor_dispatcher.py +++ b/tests/test_monitor_dispatcher.py @@ -18,6 +18,29 @@ def test_load_monitor_targets_reads_ibkr_specific_env(monkeypatch): assert load_monitor_targets() == [{"service_name": "svc-tqqq"}] +def test_load_monitor_targets_falls_back_to_generic_env(monkeypatch): + monkeypatch.delenv("IBKR_MONITOR_DISPATCH_TARGETS_JSON", raising=False) + monkeypatch.setenv( + "MONITOR_DISPATCH_TARGETS_JSON", + '{"targets":[{"service_name":"generic-service"}]}', + ) + + assert load_monitor_targets() == [{"service_name": "generic-service"}] + + +def test_load_monitor_targets_prefers_ibkr_specific_env(monkeypatch): + monkeypatch.setenv( + "IBKR_MONITOR_DISPATCH_TARGETS_JSON", + '{"targets":[{"service_name":"ibkr-service"}]}', + ) + monkeypatch.setenv( + "MONITOR_DISPATCH_TARGETS_JSON", + '{"targets":[{"service_name":"generic-service"}]}', + ) + + assert load_monitor_targets() == [{"service_name": "ibkr-service"}] + + def test_load_monitor_targets_rejects_non_object_items(): with pytest.raises(ValueError, match="target at index 1 must be an object"): load_monitor_targets('{"targets":[{"service_name":"svc-tqqq"},"bad-target"]}') From 9580858fe678e86786030b02489df343e8b00043 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 21 Jul 2026 05:47:52 +0800 Subject: [PATCH 6/6] fix: persist setup timeout reports Co-Authored-By: Codex --- main.py | 21 +++++++++++--------- tests/test_request_handling.py | 35 ++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 9 deletions(-) diff --git a/main.py b/main.py index 3aa4cd0..ea5514c 100644 --- a/main.py +++ b/main.py @@ -1112,16 +1112,21 @@ def _handle_request( log_context = build_request_log_context() report = build_execution_report(log_context, dry_run_only_override=dry_run_only_override) - strategy_plugin_signals, strategy_plugin_error = load_strategy_plugin_signals() - attach_strategy_plugin_report( - report, - signals=strategy_plugin_signals, - error=strategy_plugin_error, - ) + if report_collector is not None: + # Register the mutable report before any deadline-bound setup work so a + # timeout while loading strategy plugins still leaves a report to persist. + report_collector.append(report) execution_window = "dry_run" if dry_run_only_override else "execution" - lock_acquired = STRATEGY_RUN_LOCK.acquire(blocking=False) + lock_acquired = False deadline_exceeded = False try: + strategy_plugin_signals, strategy_plugin_error = load_strategy_plugin_signals() + attach_strategy_plugin_report( + report, + signals=strategy_plugin_signals, + error=strategy_plugin_error, + ) + lock_acquired = STRATEGY_RUN_LOCK.acquire(blocking=False) log_runtime_event( log_context, "strategy_cycle_received", @@ -1353,8 +1358,6 @@ def _handle_request( finally: if lock_acquired: STRATEGY_RUN_LOCK.release() - if report_collector is not None: - report_collector.append(report) if not deadline_exceeded and persist_report: try: if dry_run_only_override is None: diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index a995ad8..6d4c970 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -214,6 +214,41 @@ def test_handle_dry_run_deadline_from_strategy_persists_once_and_recycles(strate assert observed["recycles"] == 1 +def test_handle_dry_run_deadline_during_plugin_setup_persists_report(strategy_module, monkeypatch): + observed = {"reports": [], "recycles": 0} + + monkeypatch.setattr(strategy_module, "build_request_log_context", lambda: types.SimpleNamespace(run_id="run-001")) + monkeypatch.setattr(strategy_module, "build_execution_report", lambda *_args, **_kwargs: {"status": "pending"}) + monkeypatch.setattr( + strategy_module, + "load_strategy_plugin_signals", + lambda: (_ for _ in ()).throw( + strategy_module.RuntimeDeadlineExceeded("plugin setup deadline exceeded") + ), + ) + monkeypatch.setattr(strategy_module, "log_runtime_event", lambda *_args, **_kwargs: None) + monkeypatch.setattr( + strategy_module, + "persist_execution_report", + lambda report, **_kwargs: observed["reports"].append(dict(report)) or "/tmp/runtime-report.json", + ) + monkeypatch.setattr( + strategy_module, + "recycle_current_process_after_response", + lambda: observed.__setitem__("recycles", observed["recycles"] + 1), + ) + + with strategy_module.app.test_request_context("/dry-run", method="POST"): + body, status = strategy_module.handle_dry_run() + + assert status == 503 + assert body == "Error" + assert observed["recycles"] == 1 + assert len(observed["reports"]) == 1 + assert observed["reports"][0]["status"] == "error" + assert observed["reports"][0]["diagnostics"]["failure_category"] == "strategy_deadline" + + def test_successful_dry_run_report_timeout_does_not_flip_strategy_result(strategy_module, monkeypatch): observed = {"persist_calls": 0, "recycles": 0}