fix: harden IBKR precheck runtime dispatch

.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

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),
+        }

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