From f1864a6aa89d168d5d80fc2e36a8577f92228e3c Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 26 Aug 2026 00:20:09 +0000 Subject: [PATCH] fix(ac-0024): address 4 blockers: NUL byte in C identity, instance_key isolation, clear_warned gate, production wiring Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- .../test_reconcile_notification.py | 82 +++++++++++++++++++ .../portfolio/reconcile_notification_store.py | 2 +- trex_core/application/reconciliation.py | 21 ++++- trex_core/application/snapshot_scheduler.py | 6 ++ trex_core/runtime/cli.py | 78 +++++++++++------- trex_core/runtime/daemon.py | 6 ++ 6 files changed, 162 insertions(+), 33 deletions(-) diff --git a/tests/application/test_reconcile_notification.py b/tests/application/test_reconcile_notification.py index 5769b29..a96e7f3 100644 --- a/tests/application/test_reconcile_notification.py +++ b/tests/application/test_reconcile_notification.py @@ -839,3 +839,85 @@ def test_publish_alarm_returns_bool(): failing = _make_failing_emitter() result2 = failing.publish_alarm("test_code", "Test title") assert result2 is False + + +# --------------------------------------------------------------------------- +# Regression tests for ac-0024 blocker fixes +# --------------------------------------------------------------------------- + +def test_c_identity_key_contains_no_nul_byte(): + """_c_identity_key must not contain NUL (\\x00) — PostgreSQL TEXT bind would fail.""" + key = _c_identity_key("my-instance", "005930") + assert "\x00" not in key, f"identity key contains NUL: {key!r}" + # Must still be collision-resistant (different symbols must produce different keys) + key2 = _c_identity_key("my-instance", "005931") + assert key != key2 + # Must still be collision-resistant (different instances must produce different keys) + key3 = _c_identity_key("other-instance", "005930") + assert key != key3 + + +def test_retry_pending_resolved_ignores_other_instance_states(): + """_retry_pending_resolved must not touch pending states belonging to a different instance.""" + from trex_core.application.reconciliation import _retry_pending_resolved + + ns = InMemoryReconcileNotificationStore() + other_slot_id = "slot-other-instance" + # pending_resolved belonging to *another* instance + ns.set_state(other_slot_id, case_code="B", instance_key="other-instance", + symbol="005930", quantity_a=10, quantity_b=7, lifecycle_state="pending_resolved") + + emitter, transport = _make_emitter() + _retry_pending_resolved( + emitter, ns, + case_code="B", + instance_key=INSTANCE, # current instance — should NOT see other-instance states + active_identity_keys=set(), + ) + + # No resolved alarm published for other instance's state + codes = _alarm_codes(transport) + assert "reconcile_mismatch_resolved" not in codes + # State preserved (not consumed by wrong instance) + assert ns.get_state(other_slot_id) is not None + + +def test_b_sync_does_not_publish_resolved_when_clear_warned_fails(): + """Resolved alarm must NOT be published when clear_warned() raises (warning still active).""" + from trex_core.application.reconciliation import sync_slot_to_broker + + class FailingWarnStore(InMemoryReconcileWarningStore): + def clear_warned(self, slot_id): + raise RuntimeError("DB unavailable") + + ns = InMemoryReconcileNotificationStore() + slot = _make_slot(qty=10.0) + slot_id_str = str(slot.slot_id) + repo = _repo_with_slot(slot) + ss = SlotService(repo) + ws = FailingWarnStore() + ws.set_warned(slot_id_str, portfolio_id="pf-A", case_code="B") + ns.set_state(slot_id_str, case_code="B", instance_key=INSTANCE, + symbol="005930", quantity_a=10, quantity_b=7, lifecycle_state="notified") + + emitter, transport = _make_emitter() + result = sync_slot_to_broker( + slot_id=slot_id_str, + slot_repo=repo, + slot_service=ss, + broker=FakeBroker({"005930": 7}), # qty matches → clear attempt made + instance_key=INSTANCE, + warn_store=ws, + alarm_emitter=emitter, + notification_store=ns, + ) + + # warning_cleared is False because clear_warned raised + assert result["warning_cleared"] is False + # No resolved alarm should have been published + codes = _alarm_codes(transport) + assert "reconcile_mismatch_resolved" not in codes + # notification state untouched + state = ns.get_state(slot_id_str) + assert state is not None + assert state["lifecycle_state"] == "notified" diff --git a/trex_core/adapters/portfolio/reconcile_notification_store.py b/trex_core/adapters/portfolio/reconcile_notification_store.py index 2d752ff..fc2b0cd 100644 --- a/trex_core/adapters/portfolio/reconcile_notification_store.py +++ b/trex_core/adapters/portfolio/reconcile_notification_store.py @@ -4,7 +4,7 @@ Tracks per-identity B/C-case mismatch notification fingerprints so that duplicate alarms are suppressed across scheduler ticks and process restarts, and pending resolved transitions can be retried. -PK: identity_key (slot_id for B-case; "{instance_key}\\x00C\\x00{symbol}" for C-case). +PK: identity_key (slot_id for B-case; "{instance_key}\\x1fC\\x1f{symbol}" for C-case). lifecycle_state: 'notified' | 'pending_resolved'. """ from __future__ import annotations diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index 48fc582..42fa628 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -105,8 +105,12 @@ def _normalize_legacy_cost_basis(value: object) -> tuple[float | None, str, str # --------------------------------------------------------------------------- def _c_identity_key(instance_key: str, symbol: str) -> str: - """Stable C-case notification identity key (null-byte separated to avoid collisions).""" - return f"{instance_key}\x00C\x00{symbol}" + """Stable C-case notification identity key (PostgreSQL TEXT-safe, collision-resistant). + + Uses ASCII Unit Separator (\\x1f) which PostgreSQL TEXT accepts (only \\x00 is + forbidden) and which cannot appear in well-formed instance_key / symbol values. + """ + return f"{instance_key}\x1fC\x1f{symbol}" def _build_alarm_fields( @@ -286,12 +290,17 @@ def _retry_pending_resolved( notification_store, *, case_code: str, + instance_key: str, active_identity_keys: set[str], ) -> None: """Retry pending_resolved publishes for identities no longer actively mismatched. Called at the end of each reconcile pass to drain the pending_resolved queue for identities that have since been cleared by operator sync/verify. + + Only processes states belonging to ``instance_key`` so that a shared + PostgreSQL store used by multiple scheduler instances does not let instance A + re-publish resolved events for instance B's pending states. """ try: pending = notification_store.list_states_by_lifecycle("pending_resolved") @@ -304,6 +313,8 @@ def _retry_pending_resolved( for state in pending: if state.get("case_code") != case_code: continue + if state.get("instance_key") != instance_key: + continue # belongs to a different scheduler instance — do not touch identity_key = state["identity_key"] if identity_key in active_identity_keys: continue # still mismatched — do not retry resolved yet @@ -446,6 +457,7 @@ def reconcile_core_holdings( alarm_emitter, notification_store, case_code="B", + instance_key=instance_key, active_identity_keys=active_b_identity_keys, ) except Exception: # noqa: BLE001 @@ -568,6 +580,7 @@ def reconcile_positions( alarm_emitter, notification_store, case_code="C", + instance_key=instance_key, active_identity_keys=active_c_identity_keys, ) except Exception: # noqa: BLE001 @@ -857,8 +870,8 @@ def sync_slot_to_broker( ) except Exception: # noqa: BLE001 _log.exception("sync: failed to clear_warned for slot=%s", slot_id) - # req-0007 / ac-0024: publish B-case resolved alarm (best-effort, warning clear kept). - if alarm_emitter is not None and notification_store is not None: + # req-0007 / ac-0024: publish B-case resolved alarm only after warning is cleared. + if warning_cleared and alarm_emitter is not None and notification_store is not None: try: _notify_resolved( alarm_emitter, diff --git a/trex_core/application/snapshot_scheduler.py b/trex_core/application/snapshot_scheduler.py index dad3a68..e1d23f2 100644 --- a/trex_core/application/snapshot_scheduler.py +++ b/trex_core/application/snapshot_scheduler.py @@ -49,6 +49,7 @@ class SnapshotScheduler: instance_key: str = "default", alarm_emitter: Optional[AlarmEmitter] = None, warn_store: Optional[ReconcileWarningStore] = None, + notification_store=None, credential_fetch_enabled: bool = False, kis_reinit_callback: Optional[Callable[[BrokerCredentialPayload], None]] = None, credential_refresh: Optional[Callable[[], object]] = None, @@ -71,6 +72,8 @@ class SnapshotScheduler: self._alarm_emitter = alarm_emitter # BAT-050 / REQ-103: durable reconcile-warning store (Optional). self._warn_store = warn_store + # req-0007 / ac-0024: durable notification deduplication store (Optional). + self._notification_store = notification_store # dcy-036 / REQ-113/114/115: credential fetch opt-in + break-glass detection. # credential_fetch_enabled=False (default) → env-only, no portfolio-hub fetch. self._credential_fetch_enabled: bool = credential_fetch_enabled @@ -312,6 +315,8 @@ class SnapshotScheduler: 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") @@ -329,6 +334,7 @@ class SnapshotScheduler: 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") diff --git a/trex_core/runtime/cli.py b/trex_core/runtime/cli.py index 1d25a39..128ef21 100644 --- a/trex_core/runtime/cli.py +++ b/trex_core/runtime/cli.py @@ -707,6 +707,9 @@ def _cmd_sync(args, cfg, confirm) -> int: from trex_core.adapters.portfolio.reconcile_warning_store import ( PostgreSQLReconcileWarningStore, ) + from trex_core.adapters.portfolio.reconcile_notification_store import ( + PostgreSQLReconcileNotificationStore, + ) from trex_core.application.reconciliation import sync_slot_to_broker if not args.portfolio_id and not args.all: @@ -745,6 +748,9 @@ def _cmd_sync(args, cfg, confirm) -> int: return _broker_read_failure(exc, args.json) warn_store = PostgreSQLReconcileWarningStore(cfg.dsn) + notification_store = PostgreSQLReconcileNotificationStore(cfg.dsn) + # req-0007 / ac-0024: alarm emitter for resolved notifications. + alarm_emitter_sync, _alarm_transport_sync = _default_alarm_emitter(cfg) # Build slot_repo + slot_service from trex_core.adapters.persistence.postgresql import PostgreSQLSlotRepository @@ -789,22 +795,27 @@ def _cmd_sync(args, cfg, confirm) -> int: results = [] exit_code = 0 - for sid in slot_ids: - res = sync_slot_to_broker( - slot_id=sid, - slot_repo=slot_repo, - slot_service=slot_service, - broker=broker_port, - instance_key=instance, - warn_store=warn_store, - cost_basis_port=cost_basis_port, - dry_run=dry_run, - ) - results.append(res) - # A skipped result can still be a fail-closed live error (REQ-123). - # Do not let the no-write marker hide a failed sync from automation. - if res.get("error"): - exit_code = 1 + try: + for sid in slot_ids: + res = sync_slot_to_broker( + slot_id=sid, + slot_repo=slot_repo, + slot_service=slot_service, + broker=broker_port, + instance_key=instance, + warn_store=warn_store, + cost_basis_port=cost_basis_port, + dry_run=dry_run, + alarm_emitter=alarm_emitter_sync, + notification_store=notification_store, + ) + results.append(res) + # A skipped result can still be a fail-closed live error (REQ-123). + # Do not let the no-write marker hide a failed sync from automation. + if res.get("error"): + exit_code = 1 + finally: + _alarm_transport_sync.close() if args.json: print(_json.dumps({"instance": instance, "dry_run": dry_run, "results": results}, @@ -850,6 +861,9 @@ def _cmd_verify(args, cfg) -> int: from trex_core.adapters.portfolio.reconcile_warning_store import ( PostgreSQLReconcileWarningStore, ) + from trex_core.adapters.portfolio.reconcile_notification_store import ( + PostgreSQLReconcileNotificationStore, + ) from trex_core.application.portfolio import SnapshotStore from trex_core.application.reconciliation import verify_slot_reconcile @@ -867,6 +881,9 @@ def _cmd_verify(args, cfg) -> int: return _broker_read_failure(exc, args.json) warn_store = PostgreSQLReconcileWarningStore(cfg.dsn) + notification_store = PostgreSQLReconcileNotificationStore(cfg.dsn) + # req-0007 / ac-0024: alarm emitter for resolved notifications. + alarm_emitter_verify, _alarm_transport_verify = _default_alarm_emitter(cfg) slot_repo = PostgreSQLSlotRepository(cfg.dsn) # Build a minimal snapshot store (we only need asset_quantity for C-case check). @@ -909,18 +926,23 @@ def _cmd_verify(args, cfg) -> int: results = [] exit_code = 0 - for pid in target_pids: - res = verify_slot_reconcile( - portfolio_id=pid, - store=store, - broker=broker_port, - slot_repo=slot_repo, - instance_key=instance, - warn_store=warn_store, - ) - results.append(res) - if res["result"] == "rejected": - exit_code = 1 + try: + for pid in target_pids: + res = verify_slot_reconcile( + portfolio_id=pid, + store=store, + broker=broker_port, + slot_repo=slot_repo, + instance_key=instance, + warn_store=warn_store, + alarm_emitter=alarm_emitter_verify, + notification_store=notification_store, + ) + results.append(res) + if res["result"] == "rejected": + exit_code = 1 + finally: + _alarm_transport_verify.close() if args.json: print(_json.dumps({"instance": instance, "results": results}, diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 5c791b9..29fc8dc 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -49,6 +49,9 @@ from trex_core.adapters.portfolio import ( StubPortfolioRead, ) from trex_core.adapters.portfolio.killswitch_store import PostgreSQLKillSwitchStore +from trex_core.adapters.portfolio.reconcile_notification_store import ( + PostgreSQLReconcileNotificationStore, +) from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.account_snapshot import AccountSnapshotService from trex_core.application.broker_port import BrokerPort @@ -198,6 +201,8 @@ class Daemon: self.pause_store = PostgreSQLPauseStore(config.dsn) # BAT-050 / REQ-102: durable reconcile-warning store (slot_id PK). self.warn_store = PostgreSQLReconcileWarningStore(config.dsn) + # req-0007 / ac-0024: durable reconcile notification deduplication store. + self.notification_store = PostgreSQLReconcileNotificationStore(config.dsn) # BAT-012: day-scoped kill-switch exclusion store + service. self.killswitch_store = PostgreSQLKillSwitchStore(config.dsn) self.portfolio_read = _build_portfolio_read(config) @@ -273,6 +278,7 @@ class Daemon: instance_key=config.instance_key, alarm_emitter=self.alarm_emitter, # dcy-033 / REQ-100 warn_store=self.warn_store, # BAT-050 / REQ-103: durable reconcile-warning + notification_store=self.notification_store, # req-0007 / ac-0024 credential_fetch_enabled=config.credential_fetch_enabled, # dcy-036 / REQ-113 credential_refresh=self.broker_provider.refresh_day_start_credential, # dcy-036 / REQ-113 AC-2 / REQ-115 AC-1: KISBroker reinit callback.