feat(ops): track broker balance worker failure lifecycle and backoff

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

trex_core/application/account_snapshot.py

@@ -52,10 +52,11 @@ class AccountSnapshotUnavailable(RuntimeError):
         "read_failed",
     })
 
-    def __init__(self, code: str) -> None:
+    def __init__(self, code: str, *, detail_code: str | None = None) -> None:
         if code not in self._CODES:
             raise ValueError(f"unsupported account snapshot error code: {code}")
         self.code = code
+        self.detail_code = detail_code
         super().__init__(f"account snapshot unavailable ({code})")
 
 
@@ -217,7 +218,9 @@ class AccountSnapshotService:
             return self._broker.get_balance()
         except BrokerBalanceError as exc:
             if exc.code != "transport_failed":
-                raise AccountSnapshotUnavailable("live_read_failed") from None
+                raise AccountSnapshotUnavailable(
+                    "live_read_failed", detail_code=exc.code
+                ) from None
         snapshot = self._load_current()
         age = self._age(snapshot.holdings_captured_at, self._sell_max_age)
         _log.warning(

trex_core/application/balance_health.py

@@ -0,0 +1,139 @@
+"""Broker-balance worker health tracking with bounded backoff."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+from threading import Lock
+from typing import Callable
+
+
+@dataclass(frozen=True)
+class BalanceWorkerHealth:
+    worker: str
+    last_success_at: datetime | None
+    consecutive_failures: int
+    failure_code: str | None
+    failure_at: datetime | None
+    next_retry_at: datetime | None
+
+    def as_dict(self, *, now: datetime, stale_after_sec: float) -> dict[str, object]:
+        stale = False
+        if self.last_success_at is not None and stale_after_sec >= 0:
+            stale = (now - self.last_success_at).total_seconds() > stale_after_sec
+        state = "ok"
+        if self.last_success_at is None:
+            state = "unavailable"
+        elif self.consecutive_failures > 0:
+            state = "unavailable"
+        elif stale:
+            state = "stale"
+        return {
+            "worker": self.worker,
+            "state": state,
+            "last_success_at": (
+                self.last_success_at.astimezone(UTC).isoformat()
+                if self.last_success_at is not None
+                else None
+            ),
+            "consecutive_failures": self.consecutive_failures,
+            "failure_code": self.failure_code,
+            "failure_at": (
+                self.failure_at.astimezone(UTC).isoformat()
+                if self.failure_at is not None
+                else None
+            ),
+            "next_retry_at": (
+                self.next_retry_at.astimezone(UTC).isoformat()
+                if self.next_retry_at is not None
+                else None
+            ),
+            "stale": stale,
+        }
+
+
+class BalanceHealthTracker:
+    def __init__(
+        self,
+        *,
+        clock: Callable[[], datetime] | None = None,
+        base_backoff_sec: float = 15.0,
+        max_backoff_sec: float = 300.0,
+    ) -> None:
+        self._clock = clock or (lambda: datetime.now(tz=UTC))
+        self._base_backoff_sec = max(1.0, float(base_backoff_sec))
+        self._max_backoff_sec = max(self._base_backoff_sec, float(max_backoff_sec))
+        self._lock = Lock()
+        self._states: dict[str, BalanceWorkerHealth] = {}
+
+    def _now(self) -> datetime:
+        return self._clock()
+
+    def _get(self, worker: str) -> BalanceWorkerHealth:
+        return self._states.get(
+            worker,
+            BalanceWorkerHealth(
+                worker=worker,
+                last_success_at=None,
+                consecutive_failures=0,
+                failure_code=None,
+                failure_at=None,
+                next_retry_at=None,
+            ),
+        )
+
+    def record_success(self, worker: str, *, captured_at: datetime | None = None) -> str | None:
+        with self._lock:
+            prev = self._get(worker)
+            recovered = prev.consecutive_failures > 0
+            now = self._now()
+            self._states[worker] = BalanceWorkerHealth(
+                worker=worker,
+                last_success_at=captured_at or now,
+                consecutive_failures=0,
+                failure_code=None,
+                failure_at=None,
+                next_retry_at=None,
+            )
+        return "recovered" if recovered else None
+
+    def record_failure(self, worker: str, *, code: str) -> tuple[str | None, float]:
+        with self._lock:
+            prev = self._get(worker)
+            now = self._now()
+            failures = prev.consecutive_failures + 1
+            backoff = min(
+                self._base_backoff_sec * (2 ** max(0, failures - 1)),
+                self._max_backoff_sec,
+            )
+            self._states[worker] = BalanceWorkerHealth(
+                worker=worker,
+                last_success_at=prev.last_success_at,
+                consecutive_failures=failures,
+                failure_code=code,
+                failure_at=now,
+                next_retry_at=now + timedelta(seconds=backoff),
+            )
+            lifecycle: str | None = None
+            if prev.consecutive_failures == 0 or prev.failure_code != code:
+                lifecycle = "detected"
+            elif failures in {3, 6, 12}:
+                lifecycle = "ongoing"
+        return lifecycle, backoff
+
+    def ready(self, worker: str) -> tuple[bool, float]:
+        with self._lock:
+            state = self._get(worker)
+            now = self._now()
+            retry_at = state.next_retry_at
+            if retry_at is None:
+                return True, 0.0
+            remain = (retry_at - now).total_seconds()
+            if remain <= 0:
+                return True, 0.0
+            return False, remain
+
+    def status(self, worker: str, *, stale_after_sec: float) -> dict[str, object]:
+        with self._lock:
+            state = self._get(worker)
+            now = self._now()
+        return state.as_dict(now=now, stale_after_sec=stale_after_sec)

trex_core/application/broker_port.py

@@ -20,7 +20,7 @@ class BrokerBalanceError(RuntimeError):
 
     _CODES = frozenset({
         "token_issuance_failed", "authentication_failed", "authorization_failed",
-        "transport_failed", "malformed_response", "broker_rejected",
+        "transport_failed", "timeout", "http_5xx", "malformed_response", "broker_rejected",
         "ownership_mismatch",
     })
 

trex_core/application/execution.py

@@ -30,13 +30,16 @@ from dataclasses import dataclass, replace
 from datetime import UTC, datetime, timedelta
 
 from trex_core.application.alarm_emitter import AlarmEmitter
+from trex_core.application.balance_health import BalanceHealthTracker
 from trex_core.application.broker_port import (
     BalanceItem,
+    BrokerBalanceError,
     BrokerPort,
     CancelOutcome,
     CancelState,
     OrderRequest,
 )
+from trex_core.application.account_snapshot import AccountSnapshotUnavailable
 from trex_core.application.ledger_outbox import LedgerOutbox
 from trex_core.application.ledger_port import LedgerTrade
 from trex_core.application.ledger_skip import publish_ledger_skip_alarm
@@ -139,6 +142,8 @@ class ExecutionService:
         # account snapshot only through this injected reader. Reconciliation
         # remains wired directly to the live broker.
         self._sell_dispatch_balance_reader = balance_reader or broker.get_balance
+        self._balance_health = BalanceHealthTracker(clock=self._clock)
+        self._balance_worker = "execution"
 
     # ------------------------------------------------------------
     def run_once(self) -> ExecutionResult:
@@ -171,7 +176,22 @@ class ExecutionService:
                 # surfaces via reconcile_core_holdings (REQ-062 A). No mirror
                 # fallback by design.
                 if sell_balance is None:
-                    sell_balance = self._sell_dispatch_balance_map()
+                    ready, backoff_left = self._balance_health.ready(self._balance_worker)
+                    if not ready:
+                        r.sell_skipped_no_balance += 1
+                        _log.info(
+                            "sell dispatch deferred: broker balance backoff active "
+                            "instance_key=%s worker=%s backoff_left_sec=%.1f",
+                            self._instance_key,
+                            self._balance_worker,
+                            backoff_left,
+                        )
+                        return
+                    try:
+                        sell_balance = self._sell_dispatch_balance_map()
+                    except (AccountSnapshotUnavailable, BrokerBalanceError):
+                        r.sell_skipped_no_balance += 1
+                        return
                 bal = sell_balance.get(slot.symbol or "", 0)
                 # dcy-021 / REQ-062 (D4): full sell (slot.qty is None) => broker
                 # balance; partial sell => min(order qty, broker balance). Broker
@@ -420,10 +440,60 @@ class ExecutionService:
 
     def _sell_dispatch_balance_map(self) -> dict:
         """Balance used only to size a not-yet-submitted pending sell."""
-        return {
-            b.symbol: int(b.hldg_qty)
-            for b in self._sell_dispatch_balance_reader()
-        }
+        try:
+            holdings = self._sell_dispatch_balance_reader()
+        except AccountSnapshotUnavailable as exc:
+            code = "stale_snapshot" if exc.code == "expired" else (
+                getattr(exc, "detail_code", None) or exc.code
+            )
+            lifecycle, backoff = self._balance_health.record_failure(
+                self._balance_worker, code=code
+            )
+            if lifecycle is not None:
+                _log.warning(
+                    "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s "
+                    "failure_code=%s consecutive_failures=%d backoff_sec=%.1f",
+                    lifecycle,
+                    self._balance_worker,
+                    self._instance_key,
+                    code,
+                    self._balance_health.status(self._balance_worker, stale_after_sec=600)[
+                        "consecutive_failures"
+                    ],
+                    backoff,
+                )
+            raise
+        except BrokerBalanceError as exc:
+            lifecycle, backoff = self._balance_health.record_failure(
+                self._balance_worker, code=exc.code
+            )
+            if lifecycle is not None:
+                _log.warning(
+                    "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s "
+                    "failure_code=%s consecutive_failures=%d backoff_sec=%.1f",
+                    lifecycle,
+                    self._balance_worker,
+                    self._instance_key,
+                    exc.code,
+                    self._balance_health.status(self._balance_worker, stale_after_sec=600)[
+                        "consecutive_failures"
+                    ],
+                    backoff,
+                )
+            raise
+        lifecycle = self._balance_health.record_success(self._balance_worker)
+        if lifecycle == "recovered":
+            state = self._balance_health.status(self._balance_worker, stale_after_sec=600)
+            _log.info(
+                "broker balance recovered: worker=%s instance_key=%s last_success_at=%s",
+                self._balance_worker,
+                self._instance_key,
+                state["last_success_at"],
+            )
+        return {b.symbol: int(b.hldg_qty) for b in holdings}
+
+    def balance_worker_health(self, *, stale_after_sec: float = 600.0) -> dict[str, object]:
+        return self._balance_health.status(self._balance_worker, stale_after_sec=stale_after_sec)
 
     def _resolve_fill_price(self, fs, slot) -> tuple[float | None, bool]:
         """Resolve the fill price from FillStatus, with limit_price backfill.

trex_core/application/snapshot_scheduler.py

@@ -16,7 +16,8 @@ from datetime import date, datetime, time, timedelta, timezone
 from typing import Callable, Optional
 
 from trex_core.application.alarm_emitter import AlarmEmitter
-from trex_core.application.broker_port import BrokerPort
+from trex_core.application.balance_health import BalanceHealthTracker
+from trex_core.application.broker_port import BrokerBalanceError, BrokerPort
 from trex_core.application.portfolio import (
     BrokerCredentialPayload,
     DegradedReason,
@@ -95,6 +96,7 @@ class SnapshotScheduler:
         # REQ-114: entry gate. True = new entries allowed (default).
         # credential_fetch_enabled=False keeps this always True (no gate).
         self._entry_allowed: bool = True
+        self._balance_health = BalanceHealthTracker(clock=self._clock)
 
     # ---- callable from a daemon thread loop --------------------------------
     def run_forever(self, stop: threading.Event, *, tick_sec: float = 60.0) -> None:
@@ -310,14 +312,18 @@ class SnapshotScheduler:
         self._store.swap(items)
         try:
             # BAT-050 / REQ-103: pass warn_store + slot_repo for C-case durable warning.
-            reconcile_positions(
-                store=self._store,
-                broker=self._broker,
-                slot_repo=self._slot_repo,
-                instance_key=self._instance_key,
-                warn_store=self._warn_store,
-                alarm_emitter=self._alarm_emitter,
-                notification_store=self._notification_store,
+            self._run_balance_worker(
+                worker="position_reconciliation",
+                stale_after_sec=24 * 60 * 60,
+                run=lambda: reconcile_positions(
+                    store=self._store,
+                    broker=self._broker,
+                    slot_repo=self._slot_repo,
+                    instance_key=self._instance_key,
+                    warn_store=self._warn_store,
+                    alarm_emitter=self._alarm_emitter,
+                    notification_store=self._notification_store,
+                ),
             )
         except Exception:  # noqa: BLE001
             _log.exception("reconciliation pass failed")
@@ -330,18 +336,79 @@ class SnapshotScheduler:
                 try:
                     from trex_core.application.reconciliation import reconcile_core_holdings
                     # BAT-050 / REQ-103: pass warn_store for B-case durable warning.
-                    reconcile_core_holdings(
-                        slot_repo=self._slot_repo, slot_service=self._slot_service,
-                        broker=self._broker, instance_key=self._instance_key,
-                        alarm_emitter=self._alarm_emitter,  # dcy-033 / REQ-100
-                        warn_store=self._warn_store,  # BAT-050 / REQ-103
-                        notification_store=self._notification_store,  # req-0007 / ac-0024
+                    self._run_balance_worker(
+                        worker="core_holding_reconciliation",
+                        stale_after_sec=24 * 60 * 60,
+                        run=lambda: reconcile_core_holdings(
+                            slot_repo=self._slot_repo, slot_service=self._slot_service,
+                            broker=self._broker, instance_key=self._instance_key,
+                            alarm_emitter=self._alarm_emitter,  # dcy-033 / REQ-100
+                            warn_store=self._warn_store,  # BAT-050 / REQ-103
+                            notification_store=self._notification_store,  # req-0007 / ac-0024
+                        ),
                     )
                 except Exception:  # noqa: BLE001
                     _log.exception("core-holding divergence reconcile failed")
         self._last_fired_date = self._clock()
         return True
 
+    def _run_balance_worker(
+        self,
+        *,
+        worker: str,
+        stale_after_sec: float,
+        run: Callable[[], object],
+    ) -> None:
+        ready, backoff_left = self._balance_health.ready(worker)
+        if not ready:
+            _log.info(
+                "broker balance worker deferred: worker=%s instance_key=%s backoff_left_sec=%.1f",
+                worker,
+                self._instance_key,
+                backoff_left,
+            )
+            return
+        try:
+            run()
+        except BrokerBalanceError as exc:
+            lifecycle, backoff = self._balance_health.record_failure(worker, code=exc.code)
+            if lifecycle is not None:
+                state = self._balance_health.status(worker, stale_after_sec=stale_after_sec)
+                _log.warning(
+                    "broker balance unavailable: lifecycle=%s worker=%s instance_key=%s "
+                    "failure_code=%s consecutive_failures=%s backoff_sec=%.1f",
+                    lifecycle,
+                    worker,
+                    self._instance_key,
+                    exc.code,
+                    state["consecutive_failures"],
+                    backoff,
+                )
+            return
+        lifecycle = self._balance_health.record_success(worker)
+        if lifecycle == "recovered":
+            state = self._balance_health.status(worker, stale_after_sec=stale_after_sec)
+            _log.info(
+                "broker balance recovered: worker=%s instance_key=%s last_success_at=%s",
+                worker,
+                self._instance_key,
+                state["last_success_at"],
+            )
+
+    def balance_worker_health(
+        self,
+        *,
+        stale_after_sec: float = 24 * 60 * 60,
+    ) -> dict[str, dict[str, object]]:
+        return {
+            "position_reconciliation": self._balance_health.status(
+                "position_reconciliation", stale_after_sec=stale_after_sec
+            ),
+            "core_holding_reconciliation": self._balance_health.status(
+                "core_holding_reconciliation", stale_after_sec=stale_after_sec
+            ),
+        }
+
     # ---- asset-centric slot reconcile (dcy-020/REQ-058, BAT-028) ----------
     def _reconcile_slots(self, items: list) -> None:
         universe = {