From 8b3e7ac7530b563461faf33f27414ac5a688c5da Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sun, 13 Sep 2026 06:40:06 +0000 Subject: [PATCH] fix(broker): cap outage backoff overflow and expand worker matrix coverage Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .proofline/criteria/ac-0031.md | 4 +- tests/application/test_balance_health.py | 22 +++ tests/application/test_execution.py | 66 +++++++ tests/application/test_snapshot_scheduler.py | 171 +++++++++++++++++++ trex_core/application/balance_health.py | 18 +- 5 files changed, 275 insertions(+), 6 deletions(-) diff --git a/.proofline/criteria/ac-0031.md b/.proofline/criteria/ac-0031.md index 2b5c28b..4d3c0f0 100644 --- a/.proofline/criteria/ac-0031.md +++ b/.proofline/criteria/ac-0031.md @@ -9,7 +9,7 @@ status: draft `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 분류가 보존되어야 한다. +balance read failure는 `timeout`, `http_5xx`, `transport_failed`, `authentication_failed`, `token_issuance_failed`(credential failure), `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으로 해석되어서는 안 된다. @@ -22,6 +22,6 @@ operator-visible status surface는 instance+worker별 상태를 노출해야 하 - 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를 검증한다. +- classification/fail-closed: empty success, timeout, HTTP 5xx, auth/credential(`authentication_failed`, `token_issuance_failed`), 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/tests/application/test_balance_health.py b/tests/application/test_balance_health.py index 29be582..e7ed80a 100644 --- a/tests/application/test_balance_health.py +++ b/tests/application/test_balance_health.py @@ -23,6 +23,28 @@ def test_long_outage_backoff_is_capped_and_status_stays_valid(): assert state["failure_code"] == "http_5xx" +def test_long_outage_over_1100_failures_never_overflows_and_stays_capped(): + now = {"t": datetime(2026, 9, 12, tzinfo=UTC)} + tracker = BalanceHealthTracker(clock=lambda: now["t"], base_backoff_sec=15, max_backoff_sec=300) + + last_lifecycle = None + for i in range(1_100): + lifecycle, backoff = tracker.record_failure("execution", code="http_5xx") + last_lifecycle = lifecycle + if i >= 5: + assert backoff == 300.0 + else: + assert backoff <= 300.0 + now["t"] += timedelta(seconds=301) + + state = tracker.status("execution", stale_after_sec=600) + assert state["state"] == "unavailable" + assert state["consecutive_failures"] == 1_100 + assert state["failure_code"] == "http_5xx" + assert state["next_retry_at"] is not None + assert last_lifecycle in {None, "ongoing"} + + 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"]) diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index 9981d51..6037f9a 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -7,6 +7,8 @@ from __future__ import annotations from datetime import UTC, datetime, timedelta +import pytest + from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore from trex_core.application.account_snapshot import AccountSnapshotUnavailable @@ -303,6 +305,70 @@ def test_pending_sell_health_preserves_non_transient_snapshot_failure_code(): assert health["failure_code"] == "authentication_failed" +@pytest.mark.parametrize( + "error_code", + [ + "timeout", + "http_5xx", + "authentication_failed", + "token_issuance_failed", + "malformed_response", + ], + ids=["timeout", "http_5xx", "auth", "token", "pagination_middle_malformed"], +) +def test_execution_wrapper_classification_fail_closed_matrix(error_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") + + calls = {"n": 0} + + def fail_reader(): + calls["n"] += 1 + raise AccountSnapshotUnavailable("live_read_failed", detail_code=error_code) + + execution = ExecutionService( + svc, + repo, + broker, + instance_key=INST, + balance_reader=fail_reader, + ) + result = execution.run_once() + assert result.dispatched == 0 + assert result.sell_skipped_no_balance == 1 + assert calls["n"] == 1 + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "unavailable" + assert health["consecutive_failures"] == 1 + assert health["failure_code"] == error_code + + +def test_execution_wrapper_treats_validated_empty_success_as_success_not_failure(): + 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: [], + ) + result = execution.run_once() + assert result.dispatched == 0 + assert result.sell_skipped_no_balance == 1 # broker flat is valid success + health = execution.balance_worker_health(stale_after_sec=600) + assert health["state"] == "ok" + assert health["consecutive_failures"] == 0 + assert health["failure_code"] is None + + 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_snapshot_scheduler.py b/tests/application/test_snapshot_scheduler.py index 1ec45fd..067956f 100644 --- a/tests/application/test_snapshot_scheduler.py +++ b/tests/application/test_snapshot_scheduler.py @@ -5,6 +5,8 @@ from datetime import UTC, datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import MagicMock +import pytest + from trex_core.adapters.broker.paper import PaperBroker from trex_core.adapters.portfolio.stub_read import StubPortfolioRead from trex_core.application.alarm_emitter import AlarmEmitter @@ -896,3 +898,172 @@ def test_run_forever_retries_due_worker_on_production_loop_path(): state = sched.balance_worker_health(stale_after_sec=86_400) assert state["position_reconciliation"]["state"] == "ok" assert state["position_reconciliation"]["consecutive_failures"] == 0 + + +@pytest.mark.parametrize( + "error_code", + [ + "timeout", + "http_5xx", + "authentication_failed", + "token_issuance_failed", + "malformed_response", + ], + ids=["timeout", "http_5xx", "auth", "token", "pagination_middle_malformed"], +) +def test_position_worker_classification_matrix_is_fail_closed(error_code): + from trex_core.application.broker_port import BrokerBalanceError + from trex_core.domain.slot import SlotService + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class ReadOnce: + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=1)] + + class RaiseOrListBroker: + def __init__(self, failure: str | None): + self.failure = failure + + def get_balance(self): + if self.failure is not None: + raise BrokerBalanceError(self.failure) + return [] + + store = SnapshotStore() + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + store.swap([PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=1)]) + store.set_warning("pf-X", True) + sched = SnapshotScheduler( + read_port=ReadOnce(), + store=store, + broker=RaiseOrListBroker(error_code), + core_holdings_broker=RaiseOrListBroker(None), + slot_repo=slot_repo, + slot_service=slot_service, + ) + + assert sched.run_once() is True + health = sched.balance_worker_health(stale_after_sec=86_400) + assert health["position_reconciliation"]["state"] == "unavailable" + assert health["position_reconciliation"]["failure_code"] == error_code + assert health["position_reconciliation"]["consecutive_failures"] == 1 + assert health["core_holding_reconciliation"]["state"] == "ok" + # fail-closed on failure: warning clear path did not run. + assert store.get("pf-X").warning is True + + +def test_position_worker_validated_empty_success_does_not_create_failure_state(): + from trex_core.domain.slot import SlotService + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class ReadOnce: + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)] + + class EmptyBalanceBroker: + def get_balance(self): + return [] + + 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=ReadOnce(), + store=store, + broker=EmptyBalanceBroker(), + core_holdings_broker=EmptyBalanceBroker(), + slot_repo=slot_repo, + slot_service=slot_service, + ) + + assert sched.run_once() is True + 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["position_reconciliation"]["failure_code"] is None + + +@pytest.mark.parametrize( + "error_code", + [ + "timeout", + "http_5xx", + "authentication_failed", + "token_issuance_failed", + "malformed_response", + ], + ids=["timeout", "http_5xx", "auth", "token", "pagination_middle_malformed"], +) +def test_core_worker_classification_matrix_is_fail_closed(error_code): + from trex_core.application.broker_port import BrokerBalanceError + from trex_core.domain.slot import SlotService, SlotState + from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository + + class ReadOnce: + def whoami(self): + return "d-1" + + def get_portfolios(self, _daemon_id): + return [PortfolioItem(portfolio_id="pf-X", symbol="X", cash_balance=100, asset_quantity=0)] + + class EmptyBalanceBroker: + def get_balance(self): + return [] + + class ErrorBroker: + def __init__(self, failure: str): + self.failure = failure + + def get_balance(self): + raise BrokerBalanceError(self.failure) + + slot_repo = InMemorySlotRepository() + slot_service = SlotService(slot_repo, instance_key="default") + slot_repo.reconcile_universe("default", {"pf-X": "X"}) + slot = slot_repo.find_slot_by_portfolio("default", "pf-X") + pending = slot_service.transit_slot( + slot.slot_id, + SlotState.INACTIVE, + SlotState.PENDING_BUY, + reason="buy", + actor="test", + symbol="X", + ) + slot_service.transit_slot( + pending.slot_id, + SlotState.PENDING_BUY, + SlotState.HOLDING, + reason="filled", + actor="test", + avg_price=100.0, + qty=1.0, + ) + before = slot_repo.fetch_slot(slot.slot_id) + sched = SnapshotScheduler( + read_port=ReadOnce(), + store=SnapshotStore(), + broker=EmptyBalanceBroker(), + core_holdings_broker=ErrorBroker(error_code), + slot_repo=slot_repo, + slot_service=slot_service, + ) + + assert sched.run_once() is True + health = sched.balance_worker_health(stale_after_sec=86_400) + assert health["position_reconciliation"]["state"] == "ok" + assert health["core_holding_reconciliation"]["state"] == "unavailable" + assert health["core_holding_reconciliation"]["failure_code"] == error_code + assert health["core_holding_reconciliation"]["consecutive_failures"] == 1 + # fail-closed on failure: no slot mutation on this failed core reconcile. + after = slot_repo.fetch_slot(slot.slot_id) + assert after.status is before.status + assert after.qty == before.qty diff --git a/trex_core/application/balance_health.py b/trex_core/application/balance_health.py index 37194d5..adbbaea 100644 --- a/trex_core/application/balance_health.py +++ b/trex_core/application/balance_health.py @@ -1,6 +1,7 @@ """Broker-balance worker health tracking with bounded backoff.""" from __future__ import annotations +import math from dataclasses import dataclass from datetime import UTC, datetime, timedelta from threading import Lock @@ -64,6 +65,11 @@ class BalanceHealthTracker: self._max_backoff_sec = max(self._base_backoff_sec, float(max_backoff_sec)) self._lock = Lock() self._states: dict[str, BalanceWorkerHealth] = {} + if self._base_backoff_sec >= self._max_backoff_sec: + self._max_backoff_exponent = 0 + else: + ratio = self._max_backoff_sec / self._base_backoff_sec + self._max_backoff_exponent = int(math.floor(math.log2(ratio))) def _now(self) -> datetime: return self._clock() @@ -101,10 +107,14 @@ class BalanceHealthTracker: 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, - ) + exponent = max(0, failures - 1) + if exponent > self._max_backoff_exponent: + backoff = self._max_backoff_sec + else: + backoff = min( + self._base_backoff_sec * (2.0 ** exponent), + self._max_backoff_sec, + ) self._states[worker] = BalanceWorkerHealth( worker=worker, last_success_at=prev.last_success_at,