From 1f728f4d7dbbd56ac4172d303a6bd264456a6500 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 06:25:28 +0000 Subject: [PATCH] fix(broker): restore main wiring and add production retry cadence + status surface Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .proofline/criteria/ac-0031.md | 27 ++++ .proofline/identities.json | 2 +- tests/application/test_balance_health.py | 38 +++++ tests/application/test_execution.py | 31 ++++ .../test_slots_handler_broker_quantity.py | 84 +++++++++++ tests/application/test_snapshot_scheduler.py | 141 ++++++++++++++++++ trex_core/application/messaging_handlers.py | 75 ++++++++++ trex_core/application/snapshot_scheduler.py | 91 ++++++----- trex_core/runtime/daemon.py | 11 ++ 9 files changed, 464 insertions(+), 36 deletions(-) create mode 100644 .proofline/criteria/ac-0031.md create mode 100644 tests/application/test_balance_health.py diff --git a/.proofline/criteria/ac-0031.md b/.proofline/criteria/ac-0031.md new file mode 100644 index 0000000..77919e0 --- /dev/null +++ b/.proofline/criteria/ac-0031.md @@ -0,0 +1,27 @@ +--- +id: "ac-0031" +status: active +--- + +# Broker balance failure lifecycle observability and worker-isolated retry + +## Criterion + +`trex-core`는 broker balance snapshot을 사용하는 worker(`execution`, `position_reconciliation`, `core_holding_reconciliation`)별로 독립 상태를 유지해야 한다. 상태는 최소 `state(ok|unavailable|stale)`, `last_success_at`, `consecutive_failures`, sanitized `failure_code`, `next_retry_at`를 포함해야 하며, 한 worker의 성공/실패가 다른 worker의 상태를 덮어쓰면 안 된다. + +balance read failure는 `timeout`, `http_5xx`, `transport_failed`, `authentication_failed`, `authorization_failed`, `malformed_response`, `broker_rejected`, `ownership_mismatch`를 구분해야 하며, validated empty success(`[]`)는 failure로 취급하면 안 된다. live read transient failure 뒤 snapshot fallback이 stale/expired여도 downstream 상태에는 transient failure 분류가 보존되어야 한다. + +daily snapshot cadence와 balance-worker retry cadence는 분리되어야 한다. daily refresh가 이미 수행된 날짜에도 failure worker는 `next_retry_at` 이후에만 재시도되어야 하며(backoff 중 broker read 금지), 성공 시 lifecycle recovery와 정상 cadence로 자동 복귀해야 한다. fail-closed 경계는 유지되어 unavailable/stale snapshot에서 신규 주문/경고 해제 경로가 정상 snapshot으로 해석되어서는 안 된다. + +health tracker의 상태는 process-local로 유지되며 daemon restart 시 초기화된다. restart 이후 첫 failure는 새 lifecycle `detected`로 다시 기록되어야 한다. + +operator-visible status surface는 instance+worker별 상태를 노출해야 하며 account/token/credential/raw broker payload/예외 원문을 포함해서는 안 된다. + +## Verification + +- worker isolation: execution/position/core-holding 중 하나만 실패시켜도 다른 worker의 상태가 변경되지 않음을 검증한다. +- retry cadence: production scheduler path(`run_forever`/`_maybe_tick`)에서 daily refresh 여부와 무관하게 due worker만 backoff deadline 이후 재시도되는지 검증한다. +- backoff/lifecycle: 장기 outage에서 backoff가 cap에 고정되고, detected/ongoing/recovered lifecycle이 중복 경고 없이 전이되는지 검증한다. +- classification/fail-closed: empty success, timeout, HTTP 5xx, auth/credential, malformed, pagination-middle failure를 구분하고 각 worker의 fail-closed 결과와 health state를 검증한다. +- restart semantics: daemon restart(새 tracker 인스턴스) 후 이전 health/backoff가 초기화되고 첫 failure가 detected로 재발행되는지 검증한다. +- sanitization/visibility: operator status 응답에 instance+worker 상태 필드는 노출되지만 민감정보/원시 payload/예외 문자열은 노출되지 않음을 검증한다. diff --git a/.proofline/identities.json b/.proofline/identities.json index eb60750..7ce36f9 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, "next_line_number": 10, - "next_ac_number": 31 + "next_ac_number": 32 } diff --git a/tests/application/test_balance_health.py b/tests/application/test_balance_health.py new file mode 100644 index 0000000..29be582 --- /dev/null +++ b/tests/application/test_balance_health.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +from trex_core.application.balance_health import BalanceHealthTracker + + +def test_long_outage_backoff_is_capped_and_status_stays_valid(): + now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} + tracker = BalanceHealthTracker(clock=lambda: now["t"], base_backoff_sec=15, max_backoff_sec=300) + + backoffs: list[float] = [] + for _ in range(20): + lifecycle, backoff = tracker.record_failure("execution", code="http_5xx") + backoffs.append(backoff) + assert lifecycle in {None, "detected", "ongoing"} + now["t"] += timedelta(seconds=301) + + assert max(backoffs) == 300.0 + state = tracker.status("execution", stale_after_sec=600) + assert state["state"] == "unavailable" + assert state["consecutive_failures"] == 20 + assert state["failure_code"] == "http_5xx" + + +def test_restart_resets_process_local_health_and_re_emits_detected(): + now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} + before_restart = BalanceHealthTracker(clock=lambda: now["t"]) + before_restart.record_success("execution") + lifecycle, _ = before_restart.record_failure("execution", code="timeout") + assert lifecycle == "detected" + + after_restart = BalanceHealthTracker(clock=lambda: now["t"]) + lifecycle, _ = after_restart.record_failure("execution", code="timeout") + assert lifecycle == "detected" + state = after_restart.status("execution", stale_after_sec=600) + assert state["last_success_at"] is None + assert state["consecutive_failures"] == 1 diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index e5df5c4..9981d51 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -272,6 +272,37 @@ def test_pending_sell_balance_failure_is_fail_closed_with_backoff_and_recovery() assert health["consecutive_failures"] == 0 +def test_pending_sell_health_preserves_non_transient_snapshot_failure_code(): + broker = PaperBroker(fill_after_polls=0) + repo, svc, intents, _ = _wire(broker) + sid = _buy(intents, qty=3).slot_id + ExecutionService( + svc, + repo, + broker, + instance_key=INST, + ).run_once() + svc.sell(sid, actor="t", reason="full") + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + balance_reader=lambda: (_ for _ in ()).throw( + AccountSnapshotUnavailable( + "live_read_failed", + detail_code="authentication_failed", + ) + ), + ) + result = execution.run_once() + assert result.sell_skipped_no_balance == 1 + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "unavailable" + assert health["failure_code"] == "authentication_failed" + + def test_partial_sell_qty_is_min_intent_broker(): """dcy-021/REQ-062 (D4): partial sell dispatches min(intent qty, broker); remainder stays holding (broker-authoritative qty), cost basis preserved.""" diff --git a/tests/application/test_slots_handler_broker_quantity.py b/tests/application/test_slots_handler_broker_quantity.py index 7a8d571..1471f67 100644 --- a/tests/application/test_slots_handler_broker_quantity.py +++ b/tests/application/test_slots_handler_broker_quantity.py @@ -38,6 +38,14 @@ class _Reader: return list(self._holdings), self._captured_at +class _HealthReader: + def __init__(self, payload: dict[str, dict[str, object]]) -> None: + self.payload = payload + + def balance_health_for_slots_rpc(self) -> dict[str, dict[str, object]]: + return self.payload + + class _SlotsReader(Protocol): def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: ... @@ -61,6 +69,7 @@ class _Broker: def _holding_handler( *, broker_reader: _SlotsReader, + health_reader: _HealthReader | None = None, instance_key: str = "default", portfolio_id: str = "pf-A", symbol: str = "104620", @@ -96,6 +105,7 @@ def _holding_handler( instance_key=instance_key, snapshot_store=store, broker_quantity_reader=broker_reader, + broker_balance_health_reader=health_reader, ) return repo, handler @@ -229,3 +239,77 @@ def test_broker_quantity_reader_is_instance_bound_per_handler(): row_b = handler_b(_req_holding()).payload["slots"][0] assert row_a["qty"] == 440.0 assert row_b["qty"] == 111.0 + + +def test_slots_response_includes_instance_worker_balance_health(): + _, handler = _holding_handler( + broker_reader=_Reader( + holdings=[BalanceItem(symbol="104620", hldg_qty=440)], + captured_at=datetime(2026, 8, 24, 12, 0, tzinfo=UTC), + ), + health_reader=_HealthReader( + { + "execution": { + "state": "unavailable", + "last_success_at": "2026-08-24T11:59:00+00:00", + "consecutive_failures": 2, + "failure_code": "http_5xx", + "next_retry_at": "2026-08-24T12:00:15+00:00", + }, + "position_reconciliation": { + "state": "ok", + "last_success_at": "2026-08-24T12:00:00+00:00", + "consecutive_failures": 0, + "failure_code": None, + "next_retry_at": None, + }, + "core_holding_reconciliation": { + "state": "stale", + "last_success_at": "2026-08-24T08:00:00+00:00", + "consecutive_failures": 0, + "failure_code": None, + "next_retry_at": None, + }, + } + ), + ) + payload = handler(_req_holding()).payload + health = payload["broker_balance_health"] + assert health["execution"]["instance_key"] == "default" + assert health["execution"]["worker"] == "execution" + assert health["execution"]["failure_code"] == "http_5xx" + assert health["execution"]["consecutive_failures"] == 2 + assert health["position_reconciliation"]["state"] == "ok" + assert health["core_holding_reconciliation"]["state"] == "stale" + + +def test_slots_balance_health_surface_sanitizes_unexpected_fields_and_codes(): + _, handler = _holding_handler( + broker_reader=_Reader( + holdings=[BalanceItem(symbol="104620", hldg_qty=440)], + captured_at=datetime(2026, 8, 24, 12, 0, tzinfo=UTC), + ), + health_reader=_HealthReader( + { + "execution": { + "state": "unknown-state", + "last_success_at": "2026-08-24T11:59:00+00:00", + "consecutive_failures": "bad", + "failure_code": "account=12345678 token=secret", + "next_retry_at": "2026-08-24T12:00:15+00:00", + "raw_payload": {"secret": "leak"}, + "error": "Traceback: token", + } + } + ), + ) + execution = handler(_req_holding()).payload["broker_balance_health"]["execution"] + assert execution == { + "instance_key": "default", + "worker": "execution", + "state": "unavailable", + "last_success_at": "2026-08-24T11:59:00+00:00", + "consecutive_failures": 0, + "failure_code": None, + "next_retry_at": "2026-08-24T12:00:15+00:00", + } diff --git a/tests/application/test_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 381e5a2..1ec45fd 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -57,10 +57,12 @@ def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch): store = SnapshotStore() projected_broker = object() authoritative_broker = object() + unmanaged_store = object() captured = {} def _capture(**kwargs): captured["broker"] = kwargs["broker"] + captured["unmanaged_position_store"] = kwargs.get("unmanaged_position_store") return {} monkeypatch.setattr( @@ -74,11 +76,13 @@ def test_run_once_uses_authoritative_core_holdings_broker(monkeypatch): core_holdings_broker=authoritative_broker, slot_repo=object(), slot_service=SimpleNamespace(instance_key="inst-1"), + unmanaged_position_store=unmanaged_store, ) sched._reconcile_slots = lambda _items: None assert sched.run_once() is True assert captured["broker"] is authoritative_broker + assert captured["unmanaged_position_store"] is unmanaged_store def test_swap_preserves_warning_for_carried_over_portfolio(): @@ -755,3 +759,140 @@ def test_balance_worker_health_isolated_backoff_and_recovery(): state = sched.balance_worker_health(stale_after_sec=86_400) assert state["position_reconciliation"]["state"] == "ok" assert state["position_reconciliation"]["consecutive_failures"] == 0 + + +def test_maybe_tick_retries_due_worker_without_daily_refresh(): + from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + from trex_core.domain.slot import SlotService + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class CountingRead: + def __init__(self): + self.calls = 0 + + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + self.calls += 1 + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)] + + class SequenceBroker: + def __init__(self, steps): + self.steps = list(steps) + self.calls = 0 + + def get_balance(self): + self.calls += 1 + step = self.steps.pop(0) + if isinstance(step, Exception): + raise step + return step + + now = {"t": datetime(2026, 9, 12, 0, 0, tzinfo=UTC)} + read = CountingRead() + position_broker = SequenceBroker( + [BrokerBalanceError("http_5xx"), [BalanceItem("X", 0)]] + ) + core_broker = SequenceBroker([[BalanceItem("X", 0)]]) + store = SnapshotStore() + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + + sched = SnapshotScheduler( + read_port=read, + store=store, + broker=position_broker, + core_holdings_broker=core_broker, + slot_repo=slot_repo, + slot_service=slot_service, + schedule_kst=datetime(2026, 9, 12, 23, 59).time(), + clock=lambda: now["t"], + ) + + assert sched.run_once() is True + assert read.calls == 1 + assert position_broker.calls == 1 + assert core_broker.calls == 1 + + now["t"] += timedelta(seconds=10) + sched._maybe_tick() + assert read.calls == 1 + assert position_broker.calls == 1 # backoff window still active + assert core_broker.calls == 1 # no pending core failure, so no retry call + + now["t"] += timedelta(seconds=6) + sched._maybe_tick() + assert read.calls == 1 + assert position_broker.calls == 2 # due retry fired without daily snapshot refresh + health = sched.balance_worker_health(stale_after_sec=86_400) + assert health["position_reconciliation"]["state"] == "ok" + assert health["position_reconciliation"]["consecutive_failures"] == 0 + assert health["core_holding_reconciliation"]["state"] == "ok" + + +def test_run_forever_retries_due_worker_on_production_loop_path(): + from trex_core.application.broker_port import BalanceItem, BrokerBalanceError + + class LoopRead: + def __init__(self): + self.calls = 0 + + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + self.calls += 1 + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)] + + class SequenceBroker: + def __init__(self, steps): + self.steps = list(steps) + self.calls = 0 + + def get_balance(self): + self.calls += 1 + step = self.steps.pop(0) + if isinstance(step, Exception): + raise step + return step + + class StopAfter: + def __init__(self, waits: int, tick): + self._waits = waits + self._tick = tick + self._count = 0 + + def wait(self, _sec: float): + if self._count >= self._waits: + return True + self._tick(self._count) + self._count += 1 + return False + + now = {"t": datetime(2026, 9, 12, 0, 0, tzinfo=UTC)} + read = LoopRead() + position_broker = SequenceBroker( + [BrokerBalanceError("http_5xx"), [BalanceItem("X", 0)]] + ) + sched = SnapshotScheduler( + read_port=read, + store=SnapshotStore(), + broker=position_broker, + schedule_kst=datetime(2026, 9, 12, 23, 59).time(), + clock=lambda: now["t"], + ) + stop = StopAfter( + waits=3, + tick=lambda i: now.__setitem__("t", now["t"] + timedelta(seconds=(10 if i == 0 else 6))), + ) + + sched.run_forever(stop, tick_sec=1.0) + + # Boot refresh once + no extra daily refresh, but one due retry after backoff deadline. + assert read.calls == 1 + assert position_broker.calls == 2 + state = sched.balance_worker_health(stale_after_sec=86_400) + assert state["position_reconciliation"]["state"] == "ok" + assert state["position_reconciliation"]["consecutive_failures"] == 0 diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index f525d95..dc8f62f 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -192,6 +192,63 @@ class SlotsBrokerQuantityReader(Protocol): def holdings_for_slots_rpc(self) -> tuple[list[BalanceItem], datetime]: ... +class SlotsBrokerBalanceHealthReader(Protocol): + def balance_health_for_slots_rpc(self) -> dict[str, dict[str, object]]: ... + + +_BALANCE_HEALTH_STATES = frozenset({"ok", "unavailable", "stale"}) +_BALANCE_HEALTH_FAILURE_CODES = frozenset({ + "token_issuance_failed", + "authentication_failed", + "authorization_failed", + "transport_failed", + "timeout", + "http_5xx", + "malformed_response", + "broker_rejected", + "ownership_mismatch", + "stale_snapshot", + "account_mismatch", + "not_ready", + "expired", + "live_read_failed", + "read_failed", +}) + + +def _sanitize_balance_worker_status( + *, + instance_key: str, + worker: str, + raw: dict[str, object], +) -> dict[str, object]: + state = str(raw.get("state") or "unavailable") + if state not in _BALANCE_HEALTH_STATES: + state = "unavailable" + try: + consecutive_failures = max(0, int(raw.get("consecutive_failures", 0))) + except (TypeError, ValueError): + consecutive_failures = 0 + failure_code_raw = raw.get("failure_code") + failure_code = ( + str(failure_code_raw) + if isinstance(failure_code_raw, str) + and failure_code_raw in _BALANCE_HEALTH_FAILURE_CODES + else None + ) + last_success_at = raw.get("last_success_at") + next_retry_at = raw.get("next_retry_at") + return { + "instance_key": instance_key, + "worker": worker, + "state": state, + "last_success_at": last_success_at if isinstance(last_success_at, str) else None, + "consecutive_failures": consecutive_failures, + "failure_code": failure_code, + "next_retry_at": next_retry_at if isinstance(next_retry_at, str) else None, + } + + def make_slots_rpc_handler( slot_repo: SlotRepository, *, @@ -205,6 +262,7 @@ def make_slots_rpc_handler( max_active_slots: Optional[int] = None, strategy_state_store=None, broker_quantity_reader: Optional[SlotsBrokerQuantityReader] = None, + broker_balance_health_reader: Optional[SlotsBrokerBalanceHealthReader] = None, ): """Build the RPC handler for `trex.slots.req` (REQ-073/074/116). @@ -437,6 +495,23 @@ def make_slots_rpc_handler( "reason": reason, "message": message, } + if broker_balance_health_reader is not None: + try: + raw_health = broker_balance_health_reader.balance_health_for_slots_rpc() + except Exception: # noqa: BLE001 - status surface must stay available + raw_health = {} + health_payload: dict[str, dict[str, object]] = {} + if isinstance(raw_health, dict): + for worker, raw in raw_health.items(): + if not isinstance(worker, str) or not worker or not isinstance(raw, dict): + continue + health_payload[worker] = _sanitize_balance_worker_status( + instance_key=instance_key, + worker=worker, + raw=raw, + ) + if health_payload: + base_payload["broker_balance_health"] = health_payload if warn_store is not None: base_payload["excluded_slots"] = { "reconcile_quantity_mismatch": excluded_count, diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index 39da01e..c7a5da3 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -117,6 +117,7 @@ class SnapshotScheduler: _log.exception("snapshot scheduler tick failed") def _maybe_tick(self) -> None: + self._maybe_retry_balance_workers() now_kst = self._clock().astimezone(_KST) target = datetime.combine(now_kst.date(), self._schedule_kst, tzinfo=_KST) if now_kst < target: @@ -319,19 +320,7 @@ class SnapshotScheduler: self._store.swap(items) try: # BAT-050 / REQ-103: pass warn_store + slot_repo for C-case durable warning. - 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, - ), - ) + self._run_position_reconciliation_worker() except Exception: # noqa: BLE001 _log.exception("reconciliation pass failed") if self._slot_repo is not None: @@ -341,34 +330,66 @@ class SnapshotScheduler: _log.exception("slot reconcile pass failed") if self._slot_service is not None: try: - from trex_core.application.reconciliation import reconcile_core_holdings # BAT-050 / REQ-103: pass warn_store for B-case durable warning. -<<<<<<< HEAD - 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 - ), -======= - reconcile_core_holdings( - slot_repo=self._slot_repo, slot_service=self._slot_service, - broker=self._core_holdings_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 - unmanaged_position_store=self._unmanaged_position_store, ->>>>>>> origin/main - ) + self._run_core_holding_reconciliation_worker() except Exception: # noqa: BLE001 _log.exception("core-holding divergence reconcile failed") self._last_fired_date = self._clock() return True + def _run_position_reconciliation_worker(self) -> None: + 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, + ), + ) + + def _run_core_holding_reconciliation_worker(self) -> None: + from trex_core.application.reconciliation import reconcile_core_holdings + + 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._core_holdings_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 + unmanaged_position_store=self._unmanaged_position_store, + ), + ) + + def _has_pending_balance_failure(self, worker: str) -> bool: + status = self._balance_health.status(worker, stale_after_sec=24 * 60 * 60) + return int(status.get("consecutive_failures", 0)) > 0 + + def _maybe_retry_balance_workers(self) -> None: + if self._has_pending_balance_failure("position_reconciliation"): + try: + self._run_position_reconciliation_worker() + except Exception: # noqa: BLE001 + _log.exception("reconciliation retry pass failed") + if ( + self._slot_repo is not None + and self._slot_service is not None + and self._has_pending_balance_failure("core_holding_reconciliation") + ): + try: + self._run_core_holding_reconciliation_worker() + except Exception: # noqa: BLE001 + _log.exception("core-holding divergence retry failed") + def _run_balance_worker( self, *, diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 5091f86..78c5051 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -466,10 +466,21 @@ class Daemon: max_active_slots=self.config.max_active_slots, strategy_state_store=self.strategy_state_store, # dcy-028 C-1 broker_quantity_reader=self.account_snapshot_service, + broker_balance_health_reader=self, ) transport.consume_rpc_forever( topology.slots_queue(self.config.instance_key), handler, self.stop) + def balance_health_for_slots_rpc(self) -> dict[str, dict[str, object]]: + reconcile = self.snapshot_scheduler.balance_worker_health( + stale_after_sec=24 * 60 * 60, + ) + return { + "execution": self.execution.balance_worker_health(stale_after_sec=600), + "position_reconciliation": reconcile["position_reconciliation"], + "core_holding_reconciliation": reconcile["core_holding_reconciliation"], + } + def _strategy_state_set_loop(self) -> None: # dcy-028 / REQ-080: opaque strategy-state write RPC (entry-version CAS). transport = _build_transport(self.config)