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..6a43e15 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,126 @@ 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 os.environ.get("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") + 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( + 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..9636595 --- /dev/null +++ b/application/runtime_deadline.py @@ -0,0 +1,77 @@ +"""Bound synchronous Cloud Run work before the platform request deadline.""" + +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(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 +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) + + +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 8be540d..ea5514c 100644 --- a/main.py +++ b/main.py @@ -21,6 +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, + recycle_current_process_after_response, +) from application.runtime_strategy_adapters import ( build_runtime_strategy_adapters, fetch_yfinance_historical_candles, @@ -206,6 +211,8 @@ 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) +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 @@ -1093,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: @@ -1103,15 +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", @@ -1277,6 +1292,23 @@ def _handle_request( result=cycle_result.result, ) return (response_body if dry_run_only_override else cycle_result.result), 200 + 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, + }, + ) + raise except TimeoutError as exc: append_runtime_report_error( report, @@ -1326,14 +1358,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 and persist_report: + 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"): @@ -1457,6 +1490,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 +1502,29 @@ def _handle_monitor_dispatch(): dispatches_sent=result.get("dispatches_sent"), dispatch_results=result.get("results") or [], ) - return result, 200 + 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 + + +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": + 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), + "worker_recycle_required": recycle_required, + } @app.route("/run", methods=["POST", "GET"]) @@ -1475,14 +1532,81 @@ def handle_request(): return _route_with_runtime_error_fallback(_handle_request) +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", + ): + 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( + { + "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_required": True, + "worker_recycle_scheduled": recycle_on_timeout, + }, + ensure_ascii=False, + separators=(",", ":"), + ), + flush=True, + ) + if timeout_state is not None: + timeout_state["worker_recycle_required"] = True + 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"]) 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/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..9279056 100644 --- a/tests/test_monitor_dispatcher.py +++ b/tests/test_monitor_dispatcher.py @@ -1,6 +1,49 @@ import datetime as dt -from application.monitor_dispatcher import dispatch_due_monitor_targets +import pytest + +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_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"]}') def test_due_monitor_dispatches_selects_due_window_and_skips_disabled_target(): @@ -75,8 +118,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..6d4c970 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -153,6 +153,147 @@ 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 = {"recycles": 0} + + class ExpiredDeadline: + def __enter__(self): + raise strategy_module.RuntimeDeadlineExceeded("dry-run deadline exceeded") + + def __exit__(self, *_args): + return False + + monkeypatch.setattr(strategy_module, "enforce_runtime_deadline", lambda *_args, **_kwargs: ExpiredDeadline()) + 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 + + +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")) + 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: (_ 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, + "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["persist_calls"] == 1 + 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} + + 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 = {} @@ -223,7 +364,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 +382,89 @@ 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_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": []} diff --git a/tests/test_runtime_deadline.py b/tests/test_runtime_deadline.py new file mode 100644 index 0000000..54ca5f9 --- /dev/null +++ b/tests/test_runtime_deadline.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import time + +import pytest + +from application.runtime_deadline import ( + RuntimeDeadlineExceeded, + enforce_runtime_deadline, + recycle_current_process_after_response, +) + + +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 + + +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)] 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"