From d182464b7a4141efd035449d2814f11cf2ea8218 Mon Sep 17 00:00:00 2001 From: Justin Ko Date: Fri, 19 Jun 2026 22:39:43 +0900 Subject: [PATCH] =?UTF-8?q?feat(messaging):=20OPS-005=20=EC=9D=B4=EB=B2=A4?= =?UTF-8?q?=ED=8A=B8=20=EB=B0=9C=ED=96=89=20=EC=97=B0=EA=B2=B0=20=EC=8B=A0?= =?UTF-8?q?=EB=A2=B0=EC=84=B1=20(BAT-047)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 발행 전용 연결의 idle heartbeat 사망 후 첫 발행 유실 결함 제거. - REQ-091 발행 재시도 + 강제 재연결(_force_reconnect, is_open 불신) - REQ-092 idle keepalive(execution-loop 단일스레드 틱, process_data_events time_limit=0) - REQ-093 DLQ 가시성 + emit 실패 ERROR 경보(구조화 로그, dcy-029 재사용) - REQ-094 baseline 문서 정합(runtime-flows/operations/project-structure) 독립 opus QC PASS-WITH-NOTES → 시정 2건(ruff F401, AC-3 프로덕션 경로) 완료. 554 passed/47 skipped/회귀 0/ruff clean. NFR-1 wire 무회귀(interface-contract diff 0). BAT-047 release-candidate, REQ-091~094 Implemented. dcy-031. M2 트랙B. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5 --- .../design.md | 101 ++++ .../implementation.md | 63 +++ .../index.md | 36 ++ .../planning.md | 67 +++ .../verification.md | 131 +++++ docs/batches/index.md | 1 + docs/operations.md | 59 ++- docs/project-structure.md | 11 +- docs/runtime-flows.md | 26 +- ...ublisher-reliability-baseline-alignment.md | 2 +- ...q-091_publish-retry-and-force-reconnect.md | 2 +- .../req-092_publish-idle-keepalive.md | 2 +- ...093_dlq-visibility-and-emit-error-alert.md | 2 +- docs/srs/index.md | 9 +- .../messaging/test_publish_reliability.py | 455 ++++++++++++++++++ trex_core/adapters/messaging/in_memory.py | 3 + trex_core/adapters/messaging/rabbitmq.py | 117 ++++- trex_core/application/messaging.py | 9 + trex_core/runtime/daemon.py | 14 +- 19 files changed, 1067 insertions(+), 43 deletions(-) create mode 100644 docs/batches/bat-047_20260619_event-publisher-connection-reliability/design.md create mode 100644 docs/batches/bat-047_20260619_event-publisher-connection-reliability/implementation.md create mode 100644 docs/batches/bat-047_20260619_event-publisher-connection-reliability/index.md create mode 100644 docs/batches/bat-047_20260619_event-publisher-connection-reliability/planning.md create mode 100644 docs/batches/bat-047_20260619_event-publisher-connection-reliability/verification.md create mode 100644 tests/adapters/messaging/test_publish_reliability.py diff --git a/docs/batches/bat-047_20260619_event-publisher-connection-reliability/design.md b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/design.md new file mode 100644 index 0000000..1bd60f2 --- /dev/null +++ b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/design.md @@ -0,0 +1,101 @@ +# BAT-047 Design + +- Status: confirmed +- Batch: BAT-047 + +## Architecture Summary + +발행 신뢰성 보강은 `adapters/messaging/rabbitmq.py`의 발행 경로(`publish_event`, `_publish_dlq`)와 `runtime/daemon.py`의 `_make_event_emitter`에 국소화한다. wire 계약(AMQP 토폴로지/exchange/routing key/payload 스키마)·consume/RPC 경로(`consume_rpc_forever`, `consume_events_forever`, `call_rpc`)·`declare_topology`·슬롯 도메인·broker 집행 로직은 불변이다. + +변경 계층: +- `adapters/messaging/rabbitmq.py` (transport 어댑터 레이어) + - `publish_event`: 발행 실패 시 강제 재연결(`close` + `_connect`) 후 1회 재시도. 타임아웃 `socket_timeout`으로 상한 제어. `is_open` 맹신 제거. + - `keepalive_tick()`: `process_data_events(time_limit=0)` 호출 — 발행 전용 연결 소유 스레드에서만 호출. + - `_publish_dlq`: best-effort 실패 시 WARNING → ERROR 격상. +- `runtime/daemon.py` (runtime 레이어) + - `_make_event_emitter`: `TransportUnavailableError` 처리 WARNING → ERROR 격상, routing key + correlation 정보 포함. + - `_execution_loop`: 각 틱 직전 `event_publisher.keepalive_tick()` 호출(동일 스레드 보장). + +## Changed Areas + +- Modules: + - `trex_core/adapters/messaging/rabbitmq.py` — `publish_event`, `keepalive_tick`(신규), `_publish_dlq` + - `trex_core/runtime/daemon.py` — `_make_event_emitter`, `_execution_loop` +- Tests: + - `tests/adapters/messaging/test_publish_reliability.py` (신규) — AC 커버 + - `tests/application/test_messaging_event_publisher.py` — 기존 회귀 유지 +- Docs: + - `docs/runtime-flows.md` — 발행 재시도/keepalive 흐름, DLQ 관측 경로 추가(REQ-094) + - `docs/project-structure.md` — `keepalive_tick` 컴포넌트 기술 추가(REQ-094) + - `docs/operations.md` — DLQ 가시성/경보 운영 절차 신규(REQ-094) +- Interfaces: AMQP wire 계약 변경 없음. `interface-contract.md` 불변. + +## Key Decisions + +### OQ-1: keepalive 실행 위치 — Answered + +**결정: execution-loop 틱에 `event_publisher.keepalive_tick()` 삽입(동일 스레드 모델).** + +근거: +- pika `BlockingConnection`은 생성 스레드에서만 안전하게 I/O 호출 가능(공식 문서: "pika is not threadsafe"). +- `event_publisher`는 daemon 초기화 시 생성되어 `_execution_loop` 스레드(`trex-execution`)가 독점 소유. +- 별도 keepalive 스레드를 띄우면 그 스레드가 발행 전용 연결에 `process_data_events`를 호출 → 스레드 경계를 넘어 연결에 접근하는 것 → pika 스레드 비안전성 위반. +- execution-loop 틱은 `self.stop.wait(interval)` 주기로 이미 동기 루프 구조 → 각 틱 시작 시 `keepalive_tick()`을 먼저 호출하면 `trex-execution` 스레드 컨텍스트에서만 실행됨. +- keepalive 틱은 `process_data_events(time_limit=0)` 호출 — pika 문서에서 heartbeat frame 처리를 위한 권장 패턴. `time_limit=0`은 blocking 없이 socket에서 pending frame만 처리. +- `execution_interval_sec`(기본 1.0초) < `heartbeat=30`초의 절반(15초) → 30초 이내 I/O 보장, 서버 측 idle 종료 예방 성립. + +**별도 스레드 방식 기각 이유**: 공유 연결에 두 스레드가 동시 접근 가능 → 발행과 keepalive가 겹치면 연결 손상 또는 예외 발생. 뮤텍스로 보호해도 pika 내부 소켓 상태가 스레드 경계를 넘어 교란될 수 있어 권고되지 않음. + +### OQ-3: DLQ 가시성 구현 방식 — Answered + +**결정: 구조화 로그 기반 (dcy-029 관측성 자산 재사용).** + +근거: +- dcy-029(BAT-045/046)에서 구조화 로깅 인프라(`logging_setup.py`, `_log.error()`, correlation id, json 포맷, Loki 적재)가 이미 구축·검증됨. +- 추가 인프라 의존성(Prometheus, StatsD, 소비자 데몬)은 본 주기 범위 밖(M3 후보). +- 소비자 데몬 방식은 별도 상시 프로세스·새 큐·바인딩이 필요해 wire 계약 추가 가능성이 있어 교차 영향 사안(trex-orb-v1·trex-tsl-v1) 검토 필요 → 현 주기 제외. +- 로그 기반: `_publish_dlq` 호출 시 `_log.error("DLQ 유입: reason=%s rk=%s", reason, routing_key)`, DLQ 재발행 실패 시 `_log.error("DLQ publish failed", exc_info=True)`, `_make_event_emitter` 재시도 최종 실패 시 `_log.error("event publish failed (rk=%s)", rk, exc_info=True)`. +- 운영자는 Loki 쿼리 `{level="error"} |= "DLQ"` 또는 `|= "event publish failed"` 로 즉시 식별 가능. +- DLQ 깊이(queue depth)는 RabbitMQ 관리 UI(`http://broker:15672`) 또는 `rabbitmqctl list_queues name messages`로 조회 — 별도 구현 불요. + +**소비자 데몬 기각 이유**: 별도 프로세스 관리 복잡성 + 새 큐/바인딩 = wire 계약 파급 가능성. M3 검토로 이월. + +### NFR-2: 재시도/keepalive 타임아웃 상한 — Answered + +**결정:** +- **단일 publish 경로 전체 블록 상한: ≤ 5초** (재연결 1회 + 재시도 1회 합산). + - `_connect()` 내 `pika.BlockingConnection(params)`: `params.socket_timeout = 2.0` (기존 `self._timeout = 5.0`보다 강화 — 재연결용). + - `params.blocked_connection_timeout = 2.0`. + - `basic_publish` + `confirm_delivery` 대기: publisher confirms는 channel-level이므로 별도 타임아웃 없음(socket_timeout으로 상한됨). + - 강제 재연결 `close()` + `_connect(2.0s)` + `basic_publish` ≤ 5초 내 완료 설계. +- **keepalive 틱: `process_data_events(time_limit=0)`** — non-blocking, pending socket frame만 처리. + - 실질 지연: 수 밀리초 이내, execution-loop 지연 없음. +- **재시도 횟수: 1회** — at-least-once 의미를 최소화하고 중복 위험도 최소화. + +근거: execution_interval_sec 기본 1.0초인데 발행 경로가 5초 이상 블록되면 매도 타이밍 손상 가능. 2초 socket_timeout + 재연결 overhead ≤ 5초가 실전 허용 범위. keepalive는 non-blocking으로 루프 지연 없음. + +## Edge Cases + +1. **재시도 중 중복 발행**: at-least-once → orb/tsl 소비자 중복 수신 가능. 전이 이벤트는 `sequence_at` last-wins 적용 가능(멱등 가능 추정, OQ-2 delivery 검증 추적). +2. **재시도에서도 연결 실패**: `TransportUnavailableError`를 `_make_event_emitter`까지 전파 → ERROR 레벨 로그. 슬롯 전이 롤백 없음(NFR-3 — publish failure must not roll back). +3. **keepalive 호출 중 발행 겹침**: execution-loop 단일 스레드이므로 실제 동시성 없음. `_execution_loop`에서 `keepalive_tick()` → `run_once()` 순서적 호출. +4. **`MessagingTransport`(Protocol) 인터페이스 변경**: `keepalive_tick()` 메서드는 `RabbitMQTransport` 전용이므로 포트 인터페이스(`MessagingTransport`)에 추가하지 않는다. `daemon.py`에서 `isinstance` 체크 또는 `hasattr`로 호출하거나, `InMemoryTransport`에도 no-op 구현을 추가한다. 포트 인터페이스 확장은 wire 계약이 아니라 transport 구현 내부이므로 교차 영향 없음. + - **결정**: `MessagingTransport` 포트에 `keepalive_tick() -> None` (default no-op) 추가. 기존 구현(`InMemoryTransport`)에 no-op 추가. daemon에서 `self.event_publisher.keepalive_tick()` 직접 호출. 포트 인터페이스 추가는 transport 어댑터 내부이므로 wire 계약과 무관. +5. **`_publish_dlq` 내 keepalive**: DLQ 발행은 같은 channel에서 `_ensure_connected()` 후 수행. keepalive와 별도 경로이므로 간섭 없음. DLQ 발행은 no publisher-confirms이므로 block 없음. +6. **`close()` 후 `keepalive_tick()` 호출**: `_conn is None` 체크로 no-op. 데몬 shutdown 시 stop 이벤트 세팅 후 루프 종료 → keepalive 틱 호출 없음. + +## Architecture Impact + +- runtime-flows (발행 재시도/keepalive/DLQ 관측 경로 추가) +- project-structure (`keepalive_tick` 메서드, `MessagingTransport` 포트 확장 기술) + +## Reference Doc Update Plan (REQ-094) + +- `docs/runtime-flows.md`: + - `슬롯 이벤트 발행/구독` 흐름: 재시도/강제 재연결 경로, idle keepalive 동작, DLQ 유입 관측 경로 추가. + - `데몬 lifecycle`: keepalive 틱이 execution-loop 틱에 삽입됨(OQ-1 결정 반영). +- `docs/project-structure.md`: + - `messaging.py`(MessagingTransport 포트) 설명에 `keepalive_tick()` 추가. + - `rabbitmq.py` 설명에 발행 재시도·keepalive 기술 추가. +- `docs/operations.md`: + - 신규 섹션 "DLQ 가시성/경보": `trex.events.dlq` 유입 로그 확인 방법, RabbitMQ 관리 UI queue depth 조회, 운영자 조치 절차. diff --git a/docs/batches/bat-047_20260619_event-publisher-connection-reliability/implementation.md b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/implementation.md new file mode 100644 index 0000000..ac953bd --- /dev/null +++ b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/implementation.md @@ -0,0 +1,63 @@ +# BAT-047 Implementation + +- Status: completed +- Batch: BAT-047 + +## Plan Summary + +REQ-091~094 구현. 변경은 발행 경로에 국소화하고 wire 계약·consume/RPC 경로 불변 유지. + +1. `rabbitmq.py` — `_connect` 타임아웃 파라미터화, `_force_reconnect` 신규, `keepalive_tick` 신규, `publish_event` 재시도 로직, `_publish_dlq` ERROR 격상. +2. `application/messaging.py` — `MessagingTransport` 포트에 `keepalive_tick()` 추가 (no-op default). +3. `adapters/messaging/in_memory.py` — `keepalive_tick()` no-op 구현. +4. `runtime/daemon.py` — `_make_event_emitter` WARNING→ERROR 격상, `_execution_loop` keepalive 틱 삽입. +5. `tests/adapters/messaging/test_publish_reliability.py` — 신규 테스트 17건. +6. `docs/` — baseline 문서 정합 (REQ-094). + +## Changed Files + +### 코드 +- `trex_core/adapters/messaging/rabbitmq.py` + - `_connect(connect_timeout=None)`: 파라미터화, 기본값 `self._timeout` 유지 + - `_force_reconnect()`: 신규 — `close()` + `_connect(connect_timeout=2.0)` (NFR-2 5초 블록 예산) + - `keepalive_tick()`: 신규 — `process_data_events(time_limit=0)`, 비소유 스레드 호출 시 잠재적 오남용 방어(소유 스레드에서만 호출 규칙은 daemon.py에서 보장) + - `publish_event()`: 1차 실패 시 `_force_reconnect` + 재시도, `UnroutableError`는 재시도 없이 DLQ + - `_do_basic_publish()`: 신규 내부 헬퍼 — basic_publish 로직 추출(DRY) + - `_publish_dlq()`: WARNING → ERROR 2건(DLQ 유입 + DLQ 재발행 실패) +- `trex_core/application/messaging.py` + - `MessagingTransport` 프로토콜에 `keepalive_tick() -> None` 추가 +- `trex_core/adapters/messaging/in_memory.py` + - `InMemoryTransport.keepalive_tick()` no-op 구현 +- `trex_core/runtime/daemon.py` + - `_make_event_emitter`: WARNING → ERROR, routing key + slot_id + strategy_id 포함 + - `_execution_loop`: `self.event_publisher.keepalive_tick()` 각 틱 시작 시 호출 + +### 테스트 +- `tests/adapters/messaging/test_publish_reliability.py` (신규, 17 테스트) + +### 문서 (REQ-094) +- `docs/runtime-flows.md` — `슬롯 이벤트 발행/구독`, `데몬 lifecycle` 갱신 +- `docs/project-structure.md` — messaging.py, rabbitmq.py 기술 갱신 +- `docs/operations.md` — DLQ 가시성/경보 섹션 신규 + +## Execution Log + +1. `rabbitmq.py` — `_connect`, `_force_reconnect`, `keepalive_tick`, `publish_event`, `_do_basic_publish`, `_publish_dlq` 수정/추가. +2. `application/messaging.py` — `keepalive_tick` 포트 메서드 추가. +3. `in_memory.py` — no-op 추가. +4. `daemon.py` — `_make_event_emitter` + `_execution_loop` 수정. +5. 신규 테스트 17건 작성. +6. `python3 -m pytest tests/adapters/messaging/test_publish_reliability.py -v` → **17 passed**. +7. `python3 -m pytest tests/ --ignore=...rabbitmq_transport... --ignore=...rabbitmq_intent_e2e... -q` → **554 passed, 40 skipped** (기존 537 + 신규 17, 회귀 0). +8. baseline 문서 정합 (runtime-flows, project-structure, operations). + +## Validation (중간) + +- 신규 17건 전부 통과. +- 기존 537 passed → 554 passed (신규 17 추가). 회귀 0. +- `TREX_RABBITMQ_TEST_URL` 필요 통합 테스트 2건(test_rabbitmq_transport.py, test_rabbitmq_intent_e2e.py): 환경 없어 skip — 기존과 동일. + +## Remaining Risks + +- R-3(Medium): at-least-once 재시도 — 중복 이벤트 발행 가능. OQ-2(orb/tsl 소비자 멱등성)는 delivery 검증으로 추적. trex-core 범위 밖. +- R-live: 실제 pika 연결 환경(TREX_RABBITMQ_TEST_URL)에서의 keepalive/재연결 동작은 통합 테스트로만 검증 가능 — CI 환경 없어 미검증. unit mock 기반으로 logic을 검증했으나 실제 broker half-open 재현은 운영 환경 배포 후 확인 필요. diff --git a/docs/batches/bat-047_20260619_event-publisher-connection-reliability/index.md b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/index.md new file mode 100644 index 0000000..836ba35 --- /dev/null +++ b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/index.md @@ -0,0 +1,36 @@ +# BAT-047: 이벤트 발행 연결 신뢰성 — idle keepalive + 재시도/재연결 + DLQ 가시성 + baseline 정합 (OPS-005) + +- Status: release-candidate +- Profile: standard +- Source Discovery: dcy-031 +- Included REQ: REQ-091, REQ-092, REQ-093, REQ-094 +- Excluded REQ: 없음 + +## Summary +- OPS-005 운영 리포트 기반 발행 신뢰성 4건을 단일 batch로 묶는다. 발행 재시도+강제 재연결(REQ-091), idle keepalive(REQ-092), DLQ 가시성+emit ERROR 경보(REQ-093)는 같은 발행 경로(`publish_event`/`_publish_dlq`/`_make_event_emitter`)를 공유하고, 동일 messaging 검증 스위트에서 검증된다. baseline 문서 정합(REQ-094)은 구현 완료 후 같은 batch에서 처리한다. +- M2(trex-tsl-v1 라이브, 6/28 목표) 선행 조건. idle 60초 후 첫 슬롯 전이 이벤트 유실 결함 제거. + +## Scope +- In Scope: + - `trex_core/adapters/messaging/rabbitmq.py` — `publish_event` 재시도+강제 재연결(REQ-091), keepalive `process_data_events` 틱(REQ-092), `_publish_dlq` ERROR 로그(REQ-093) + - `trex_core/runtime/daemon.py` — `_make_event_emitter` ERROR 경보 강화(REQ-093), keepalive 틱 연동(REQ-092) + - `tests/adapters/messaging/` — half-open 시뮬레이션, 재시도 소진, idle keepalive, DLQ 경보 신규 테스트 + - `tests/application/test_messaging_event_publisher.py` — 기존 회귀 유지 + - `docs/runtime-flows.md`, `docs/project-structure.md`, `docs/operations.md` — baseline 정합(REQ-094) +- Out of Scope: + - 이벤트 영속 outbox(옵션 C) — M3 후보 + - consumer(orb/tsl) idempotency 구현 + - RPC client(`call_rpc`) 연결 정책 + - `docs/interface-contract.md` 변경 — AMQP wire 계약 불변 + +## Documents +- Planning: [planning.md](./planning.md) +- Design: [design.md](./design.md) +- Implementation: [implementation.md](./implementation.md) +- Verification: [verification.md](./verification.md) + +## Notes +- R-1(High): M1 라이브 trex-core — 변경은 발행 경로에 국소화, wire 계약·consume/RPC 경로 불변. opus qc-reviewer 강제(dcy-031 §10 CONFIRM①). +- R-2(Medium): pika 스레드 비안전성 — keepalive I/O는 발행 연결 소유 스레드에서만 실행(OQ-1 설계 확정). +- R-3(Medium): 재시도 at-least-once 중복 가능 — OQ-2는 delivery 검증으로 추적(소비자 멱등성). +- NFR-2 타임아웃 수치는 design 단계에서 확정. diff --git a/docs/batches/bat-047_20260619_event-publisher-connection-reliability/planning.md b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/planning.md new file mode 100644 index 0000000..7e83341 --- /dev/null +++ b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/planning.md @@ -0,0 +1,67 @@ +# BAT-047 Planning + +- Status: confirmed +- Batch: BAT-047 +- Profile: standard + +## Included REQ + +- REQ-091: 이벤트 발행 재시도 + 강제 재연결 (idle 후 첫 발행 무유실) +- REQ-092: 발행 전용 연결 idle keepalive (서버 측 heartbeat 사망 예방) +- REQ-093: DLQ 가시성/경보 + emit 실패 경보 (trex.events.dlq 관측성) +- REQ-094: 이벤트 발행 신뢰성 baseline 문서 정합 (runtime-flows / project-structure / operations 갱신) + +## Delivery Plan + +구현 순서는 의존성과 위험도 기준으로 아래와 같다. + +1. **REQ-092 keepalive 설계 확정(OQ-1)**: draft-batch-design에서 pika 스레드 소유 모델 기준으로 keepalive 실행 위치(별도 스레드 vs. execution-loop 틱) 결정. 이 결정이 REQ-092 구현 전제. + +2. **REQ-093 DLQ 가시성 방식 결정(OQ-3)**: draft-batch-design에서 dcy-029 로깅 자산 재사용 가능성 검토 후 메트릭/로그/소비자 중 택1. 이 결정이 REQ-093 DLQ 유입 관측 구현 전제. + +3. **REQ-093 NFR-2 타임아웃 수치 확정**: design에서 단일 publish 경로 전체 블록 상한 수치 결정. + +4. **구현(run-batch-implementation)**: 설계 결정 순서대로 + - Phase A: `rabbitmq.py` `publish_event` 재시도+강제 재연결(REQ-091) + `_publish_dlq` ERROR 로그(REQ-093) + - Phase B: keepalive 구현(REQ-092) — OQ-1 결정 결과 적용 + - Phase C: `daemon.py` `_make_event_emitter` ERROR 경보 강화(REQ-093) + - Phase D: 신규 테스트 작성 (half-open/재시도/keepalive/DLQ/emit ERROR AC 커버) + +5. **REQ-094 baseline 문서 정합**: 구현 완료 후 `runtime-flows.md`, `project-structure.md`, `operations.md` 갱신. + +선행 조건: REQ-091~093은 코드 범위가 분리되어 있어 병렬 구현 가능하나, keepalive(REQ-092)와 emit 경보(REQ-093)는 OQ-1/OQ-3 설계 결정이 전제여서 설계 완료 후 시작. + +## Dependencies + +- Internal: + - `trex_core/adapters/messaging/rabbitmq.py` — 변경 대상 핵심 파일 (BAT-007에서 도입됨) + - `trex_core/runtime/daemon.py` — `_make_event_emitter` 변경 대상 + - `tests/adapters/messaging/` — 기존 messaging 회귀 스위트(현재 537 passed) 전체 통과 유지 필수 + - `tests/application/test_messaging_event_publisher.py` — 기존 회귀 유지 + - dcy-029(BAT-045/046) — 구조화 로깅 인프라 자산(OQ-3 재사용 검토 대상) +- External: + - pika BlockingConnection API(`process_data_events`, `close`, `BlockingConnection`) — 버전 고정 (pyproject.toml 기준) + - RabbitMQ broker — 통합 테스트는 `TREX_RABBITMQ_TEST_URL` skip(CI 환경 없으면 skip, unit은 mock 기반) + +## Design Gate + +- Design Required Before Implementation: yes +- Reason: + - OQ-1(keepalive 실행 위치): pika 스레드 비안전성이 명시적 스레드 소유 모델 결정을 요구한다. 설계 없이 구현하면 다른 스레드에서 I/O 호출 시 연결 손상 리스크가 있다. + - OQ-3(DLQ 가시성 방식): 메트릭/로그/소비자 데몬 중 선택이 구현 구조를 결정한다. 특히 소비자 데몬 방식이면 별도 스레드/모듈이 추가된다. + - NFR-2(타임아웃 수치): execution-loop 블록 상한 수치가 코드에 상수로 박힌다. 설계에서 확정해야 AC-3 기준이 명확해진다. + - architecture impact가 `runtime-flows`(keepalive 틱, 재시도 경로)이므로 standard profile design이 필요하다. + +## Milestones + +1. draft-batch-design 완료 — OQ-1(keepalive 위치), OQ-3(DLQ 방식), NFR-2(타임아웃 수치) 확정 +2. run-batch-implementation 완료 — 신규 테스트 포함, 기존 회귀 537 passed 유지 +3. draft-batch-verification 완료 — AC별 커버 확인, baseline 문서 정합 확인 +4. QC reviewer 검증 대기 (confirm-batch-verification은 팀장 게이트) + +## Risks + +- R-1(High): M1 라이브 trex-core — wire 계약·consume/RPC 경로 회귀 시 orb(라이브)·tsl(M2)에 즉시 영향. 완화: 변경을 `publish_event`/`_publish_dlq`/`_make_event_emitter` 경로에 국소화. `declare_topology`/`consume_rpc_forever`/`consume_events_forever`/`call_rpc` 경로 불변. 기존 회귀 스위트 전체 통과 강제. +- R-2(Medium): pika 스레드 비안전성 — keepalive가 발행 전용 연결 소유 스레드 밖에서 실행되면 연결 손상. 완화: OQ-1을 design에서 확정, AC-2/AC-3(스레드 안전성) 테스트로 검증. +- R-3(Medium): 재시도 at-least-once — 중복 이벤트 발행 가능. 완화: OQ-2는 delivery 검증(소비자 orb/tsl 멱등성 확인)으로 추적. 전이 이벤트는 멱등 적용 가능 추정. +- R-4(Low): keepalive `process_data_events(time_limit=0)` 틱이 execution-loop와 같은 스레드에서 실행될 경우 루프 지연 유발 가능. 완화: NFR-2 타임아웃 수치를 design에서 확정, AC-4(블록 상한) 테스트로 검증. diff --git a/docs/batches/bat-047_20260619_event-publisher-connection-reliability/verification.md b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/verification.md new file mode 100644 index 0000000..49a8775 --- /dev/null +++ b/docs/batches/bat-047_20260619_event-publisher-connection-reliability/verification.md @@ -0,0 +1,131 @@ +# BAT-047 Verification + +- Status: approved (confirm-batch-verification, 팀장 2026-06-19) +- Batch: BAT-047 +- Verified By: trex-core-dev (draft-batch-verification → QC PASS-WITH-NOTES 시정 후 재검증) +- Verified At (KST): 2026-06-19 (최초), 2026-06-19 (QC 시정 재검증) +- Confirmed By: 팀장 (독립 opus qc-reviewer PASS-WITH-NOTES + 시정 2건 완료 + 팀장 직접 재확인) + +## Human Approval Memo (팀장 confirm-batch-verification) +- 독립 opus qc-reviewer 판정 = PASS-WITH-NOTES. 반환 2건 모두 시정 완료: + - ① ruff F401 3건 제거 → 팀장 직접 재실행 `All checks passed!` 확인. + - ② REQ-093 AC-3 테스트가 프로덕션 `Daemon._make_event_emitter()`를 실제 호출하도록 재작성 → 팀장 코드 확인(L398 `object.__new__(Daemon)` → L404 `_make_event_emitter()` → L422 `emit(...)`). + - 팀장 직접 전체 재실행: **554 passed / 47 skipped / 0 실패** (회귀 0). +- 수용한 잔여 한계(blocker 아님, release 노트에 명시할 것): + - half-open / broker 재시작 / idle 60초 생존은 mock unit으로만 검증. 실 broker 통합(`TREX_RABBITMQ_TEST_URL`)은 환경 미설정으로 skip → M1 운영 또는 broker 포함 CI로 통합검증 위임. + - NFR-2 "publish 경로 블록 ≤5초"는 재연결(2.0s)+재시도(2.0s) 구간에 엄밀 성립. 1차 실패 시도가 원 socket_timeout(5.0s)에 묶이면 최악 케이스 초과 가능 → 필요 시 후속 REQ로 초기 publish timeout 강화. +- NFR-1(M1 라이브 무회귀): wire 계약·consume/RPC·토폴로지·`interface-contract.md`(diff 0) 불변 확인 — qc-reviewer 코드 추적 + 팀장 확인. + +## Test Run Summary + +> **정정 (QC PASS-WITH-NOTES 시정 후)**: 최초 verification.md에서 "ruff clean"으로 기재했으나, QC 독립 검증에서 ruff F401 3건(threading, unittest.mock.call, trex_core.runtime.daemon.Daemon)이 검출됐다. 또한 REQ-093 AC-3 테스트가 프로덕션 `_make_event_emitter` 경로를 호출하지 않고 emit 클로저를 테스트 본문에 직접 복제하는 구조였다. 이하는 시정 후 실제 재실행 결과다. + +### 시정 후 재실행 결과 + +``` +명령: python3 -m ruff check . +결과: All checks passed! +``` + +``` +명령: python3 -m pytest tests/ --ignore=tests/adapters/messaging/test_rabbitmq_transport.py \ + --ignore=tests/adapters/messaging/test_rabbitmq_intent_e2e.py -q +결과: 554 passed, 40 skipped in 1.51s +기존 baseline: 537 passed (BAT-046 기준) +신규 추가: 17 passed (tests/adapters/messaging/test_publish_reliability.py) +회귀: 0건 +``` + +## Acceptance Criteria Mapping + +### REQ-091: 이벤트 발행 재시도 + 강제 재연결 + +| AC | 내용 | Evidence | Result | +|---|---|---|---| +| AC-1 | half-open → 재연결+재시도 → 발행 성공 | `TestPublishRetryOnHalfOpen::test_half_open_first_attempt_fails_retry_succeeds` PASSED | PASS | +| AC-2 | 재시도 소진 → `TransportUnavailableError` 전파 | `TestPublishRetryExhausted::test_retry_exhausted_raises_transport_unavailable` PASSED | PASS | +| AC-3 | 타임아웃 상한 확인 (`connect_timeout=2.0`) | `TestForceReconnectTimeout::test_force_reconnect_uses_2s_timeout` PASSED | PASS | +| AC-4 | 정상 발행(연결 살아 있음) — 재시도 경로 미호출 | `TestNormalPublishNoRetry::test_normal_publish_does_not_call_force_reconnect` PASSED | PASS | +| AC-5 | consume/RPC 경로 무회귀 | `TestConsumePaths::test_rabbitmq_transport_has_consume_methods` PASSED + 전체 회귀 554 passed 0 회귀 | PASS | +| AC-6 | broker 재시작 후 발행 복구 | `TestBrokerRestartScenario::test_broker_restart_publish_recovers` PASSED | PASS | + +### REQ-092: 발행 전용 연결 idle keepalive + +| AC | 내용 | Evidence | Result | +|---|---|---|---| +| AC-1 | idle 후 연결 생존 (keepalive가 `process_data_events` 호출) | `TestKeepaliveTick::test_keepalive_tick_calls_process_data_events_on_live_connection` PASSED | PASS | +| AC-2 | 발행 소유 스레드에서만 I/O (design 결정: execution-loop 틱에 삽입) | OQ-1 design 문서에 Answered 기록. daemon.py `_execution_loop` 내 `keepalive_tick()` 호출 코드 diff 확인. pika 스레드 경계 불침범 | PASS | +| AC-3 | 동시 발행+keepalive 충돌 없음 | `TestKeepaliveAndPublishSameThread::test_keepalive_then_publish_sequential_no_error` PASSED (단일 스레드 순차 호출) | PASS | +| AC-4 | keepalive 비블록 (`time_limit=0`) | `TestKeepaliveTimeLimitZero::test_keepalive_uses_time_limit_zero` PASSED | PASS | +| AC-5 | consume/RPC 경로 무회귀 | `TestKeepaliveDoesNotAffectConsumePaths::test_keepalive_tick_not_present_on_consume_rpc_forever_signature` PASSED | PASS | +| AC-6 | OQ-1 종료 조건 (design 문서에 keepalive 위치 결정) | `design.md` "OQ-1: Answered — execution-loop 틱 삽입" 기록 완료 | PASS | + +### REQ-093: DLQ 가시성/경보 + emit 실패 경보 + +| AC | 내용 | Evidence | Result | +|---|---|---|---| +| AC-1 | unroutable → DLQ 유입 + 관측 지표 노출 | `TestDlqVisibilityOnUnroutable::test_unroutable_calls_publish_dlq_and_logs_error` PASSED + `test_publish_dlq_logs_error_on_entry` PASSED | PASS | +| AC-2 | DLQ 깊이 파악 가능 | 운영자 RabbitMQ 관리 UI + `rabbitmqctl list_queues` 조회 절차 `docs/operations.md` §4에 기재 | PASS | +| AC-3 | emit 최종 실패 → ERROR 로그(rk+slot_id+strategy 포함) | `TestEmitErrorOnFinalPublishFailure::test_emit_logs_error_on_transport_unavailable` PASSED — **시정 후**: 복제 클로저 제거, `object.__new__(Daemon)`으로 stub 생성 후 `event_publisher`만 주입해 실제 `Daemon._make_event_emitter()`가 반환한 emit 클로저 직접 호출로 재작성. 프로덕션 경로 회귀 가드 확립. | PASS | +| AC-4 | `_publish_dlq` 재발행 실패 → ERROR 로그 | `TestDlqPublishFailureLogsError::test_dlq_publish_failure_logs_error_not_warning` PASSED | PASS | +| AC-5 | OQ-3 종료 조건 (design 문서에 DLQ 가시성 방식 결정) | `design.md` "OQ-3: Answered — 구조화 로그 기반 (dcy-029 자산 재사용)" 기록 완료 | PASS | +| AC-6 | consume/RPC 경로 무회귀 | `TestConsumePaths` PASSED + 전체 회귀 554 passed 0 회귀 | PASS | + +### REQ-094: baseline 문서 정합 + +| AC | 내용 | Evidence | Result | +|---|---|---|---| +| AC-1 | `runtime-flows.md` 발행/keepalive 경로 정합 | `docs/runtime-flows.md` — `슬롯 이벤트 발행/구독` 재시도/재연결/keepalive/DLQ 관측 경로 기술 추가 확인 | PASS | +| AC-2 | OQ-1 결정(keepalive 위치) `runtime-flows.md` 반영 | `데몬 lifecycle` flow에 `trex-execution` 틱 시작 시 `keepalive_tick()` 호출 기술 추가 확인 | PASS | +| AC-3 | OQ-3 결정(DLQ 가시성) `operations.md` DLQ 절차 신규 기재 | `docs/operations.md` §4 "DLQ 가시성 / 이벤트 발행 경보" 섹션 신규 생성 확인 | PASS | +| AC-4 | `project-structure.md` 신규 컴포넌트 반영 | `messaging.py`(keepalive_tick), `rabbitmq.py`(재시도+keepalive), `daemon.py`(keepalive/ERROR 격상), 신규 테스트 파일 기술 추가 확인 | PASS | +| AC-5 | `interface-contract.md` 불변 | diff 확인 — 본 batch에서 `docs/interface-contract.md` 변경 없음 (AMQP 토폴로지·큐·exchange·라우팅 키 불변) | PASS | + +## Wire 계약 무회귀 검증 (NFR-1) + +- `declare_topology` 경로: 변경 없음. +- `consume_rpc_forever` / `consume_events_forever` / `call_rpc` 경로: 변경 없음. +- `EVENTS_EXCHANGE`, `EVENTS_DLQ_EXCHANGE`, `EVENTS_DLQ_QUEUE` 토폴로지: 변경 없음. +- 발행 routing key 형식(`{event_type}.{strategy_id}.{symbol}`): 변경 없음. +- 페이로드 스키마: 변경 없음. +- `docs/interface-contract.md`: 변경 없음. + +## baseline 문서 동기화 (Architecture Impact: runtime-flows) + +- `docs/runtime-flows.md` — 갱신 완료 (Last Updated: 2026-06-19 BAT-047) +- `docs/project-structure.md` — 갱신 완료 (Last Updated: 2026-06-19 BAT-047) +- `docs/operations.md` — DLQ 절차 신규 기재 완료 + +## 미검증 항목 (Blocking Issues 없음, 잔여 위험으로 분류) + +- **실제 broker half-open 재현**: `TREX_RABBITMQ_TEST_URL` 환경이 없어 통합 테스트(`test_rabbitmq_transport.py`, `test_rabbitmq_intent_e2e.py`) skip됨 (기존 skip과 동일 조건). 신규 테스트는 모두 mock 기반 unit test로 로직을 검증함. +- **OQ-2 소비자 멱등성**: orb/tsl 재시도 중복 이벤트 수신 시 멱등성 확인은 각 repo 별도 추적(본 batch 밖). trex-core 범위 완료. + +## QC PASS-WITH-NOTES 시정 내역 (2026-06-19 재검증) + +QC opus 검증에서 지적된 2건을 시정했다. + +**시정 1 — ruff F401 3건 제거** +- `tests/adapters/messaging/test_publish_reliability.py` L28: `import threading` 제거 +- `tests/adapters/messaging/test_publish_reliability.py` L29: `from unittest.mock import MagicMock, call, patch` → `call` 제거 +- `trex_core.runtime.daemon.Daemon` import는 시정 2에서 실제 사용하도록 재작성해 F401 해소 + +**시정 2 — REQ-093 AC-3 테스트를 프로덕션 경로로 재작성** +- 구: 테스트 본문에 emit 클로저 직접 복제 (프로덕션 `_make_event_emitter` 미호출) +- 신: `object.__new__(Daemon)`으로 stub 인스턴스 생성 → `event_publisher`만 mock 주입 → `daemon_stub._make_event_emitter()`가 반환한 emitter를 그대로 호출 +- 효과: `daemon.py` L247-253의 `_log.error(...)` 경로를 실제로 타므로, 해당 코드가 제거/변경되면 테스트가 실패해 회귀를 가드함 + +**재검증 실제 결과** +- `python3 -m ruff check .` → `All checks passed!` +- `python3 -m pytest tests/ --ignore=... -q` → `554 passed, 40 skipped in 1.51s`, 회귀 0 + +## Ready for confirm-batch-verification + +- 신규 테스트: 17 passed +- 전체 테스트: 554 passed, 40 skipped (기존 대비 +17, 회귀 0) +- ruff: clean (F401 3건 시정 완료, 실제 실행 결과) +- AC 커버: REQ-091 AC-1~6, REQ-092 AC-1~6, REQ-093 AC-1~6, REQ-094 AC-1~5 전부 PASS 또는 evidence 있음 +- wire 계약 무회귀: PASS +- baseline 문서 정합: PASS +- Blocking Issues: 없음 +- Ready for Release: false (멈춤 — confirm-batch-verification은 팀장 게이트) diff --git a/docs/batches/index.md b/docs/batches/index.md index 41aabaa..d721990 100644 --- a/docs/batches/index.md +++ b/docs/batches/index.md @@ -54,3 +54,4 @@ | BAT-044 | standard | released | REQ-083, REQ-070 (CHG-20260616-01) | dcy-030 (issue #28) | [bat-044](./bat-044_20260616_slippage-tick-size-snapping/) | | BAT-045 | standard | released | REQ-084, REQ-085, REQ-087 | dcy-029 (issue #29, P0) | [bat-045](./bat-045_20260616_observability-logging-foundation/) | | BAT-046 | standard | released | REQ-086, REQ-088, REQ-089, REQ-090 | dcy-029 (issue #29, P1) | [bat-046](./bat-046_20260616_observability-classification-pnl-runbook/) | +| BAT-047 | standard | release-candidate | REQ-091, REQ-092, REQ-093, REQ-094 | dcy-031 | [bat-047](./bat-047_20260619_event-publisher-connection-reliability/) | diff --git a/docs/operations.md b/docs/operations.md index e15858f..e0759d8 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -94,7 +94,62 @@ grep "event=order_reject" /tmp/trex-core.log # 거부 — ERROR 레벨 --- -## 4. 디버깅 / 트러블슈팅 +## 4. DLQ 가시성 / 이벤트 발행 경보 (REQ-093 / dcy-031 / BAT-047) + +trex-core는 `trex.events.dlq`(durable) 큐와 구조화 로그를 통해 이벤트 발행 실패를 운영자에게 노출한다. + +### DLQ 유입 원인 + +| 원인 | 경로 | 로그 | +|---|---|---| +| 구독자 없는 이벤트(unroutable) | `UnroutableError` → `_publish_dlq(reason="unroutable")` | ERROR `DLQ entry: reason=unroutable rk=...` | +| DLQ 재발행 자체 실패 | `_publish_dlq` 내 exception | ERROR `DLQ publish failed (rk=...)` | +| 발행 최종 실패(재연결+재시도 후) | `_make_event_emitter` → `TransportUnavailableError` | ERROR `event publish failed (rk=... slot_id=... strategy=...)` | + +### 로그 기반 경보 (Loki / grep) + +```bash +# DLQ 유입 확인 +grep '"level":"error"' /tmp/trex-core.log | grep "DLQ" +# 또는 text 포맷 +grep "ERROR.*DLQ" /tmp/trex-core.log + +# 발행 최종 실패 확인 +grep "event publish failed" /tmp/trex-core.log +``` + +Loki LogQL: +```logql +{job="trex-core"} | json | level="error" | line_contains "DLQ" +{job="trex-core"} | json | level="error" | line_contains "event publish failed" +``` + +### RabbitMQ queue depth 확인 + +```bash +# rabbitmqctl로 DLQ 깊이 조회 +rabbitmqctl list_queues name messages | grep dlq + +# RabbitMQ 관리 UI: http://:15672 → Queues → trex.events.dlq +``` + +### 운영자 조치 절차 + +1. **DLQ 유입 경보 수신**: 로그 `ERROR ... DLQ entry` 발견. +2. **원인 분류**: + - `reason=unroutable`: 해당 strategy_id의 전략 데몬(orb/tsl)이 꺼져 있거나 큐 바인딩이 없는 경우. 전략 데몬 기동 상태 확인. + - `DLQ publish failed`: RabbitMQ 자체 연결 문제. broker 상태 확인. + - `event publish failed`: 발행 연결 재연결+재시도 후에도 실패. broker 재시작 또는 네트워크 장애. trex-core 재기동 검토. +3. **DLQ 내용 소비**(선택): 적체된 메시지는 수동 drain 또는 strategy 재기동 후 catch-up. +4. **임계값 기준**: DLQ depth > 0 = 경보 대상. depth > 100 = 즉시 조치 권고. + +### idle heartbeat 예방 (REQ-092) + +trex-core는 발행 연결의 idle 종료를 예방하기 위해 `trex-execution` 루프 각 틱(기본 1초)마다 `event_publisher.keepalive_tick()`(`process_data_events(time_limit=0)`)을 호출한다. `heartbeat=30s` 설정 대비 충분한 I/O 주기를 보장한다. + +--- + +## 5. 디버깅 / 트러블슈팅 | 증상 | 원인 | 조치 | |---|---|---| @@ -118,6 +173,8 @@ psql "$TREX_POSTGRESQL_DSN" -c "SELECT slot_id, from_state, to_state, reason, ac --- +--- + ## 관련 문서 - `docs/interface-contract.md` — RabbitMQ 토폴로지 + 메시지 payload + reason/actor 어휘 - `docs/runtime-flows.md` — 흐름 다이어그램(스냅샷/intent RPC/이벤트/집행 루프/로깅) diff --git a/docs/project-structure.md b/docs/project-structure.md index bddd3d3..5d7d27e 100644 --- a/docs/project-structure.md +++ b/docs/project-structure.md @@ -4,7 +4,7 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-10 (BAT-031 — dcy-021 포지션 사이징) +- Last Updated (KST): 2026-06-19 (BAT-047 — dcy-031 이벤트 발행 연결 신뢰성) - Source Discovery / Batch: bootstrap-baseline (pre-Discovery, 3-phase lifecycle 밖의 bootstrap 산출물) ## Project Summary @@ -74,7 +74,7 @@ │ │ ├── execution.py # ExecutionService (dispatch·poll·transition·reconcile + ledger enqueue + event emit hook) │ │ ├── ledger_port.py # LedgerPort + LedgerTrade/LedgerWriteResult/예외 (BAT-006) │ │ ├── ledger_outbox.py # LedgerOutbox Protocol + LedgerWriter + InMemoryLedgerOutbox (BAT-006) -│ │ ├── messaging.py # MessagingTransport Protocol + DTO + TransportHealth (BAT-007) +│ │ ├── messaging.py # MessagingTransport Protocol + DTO + TransportHealth + keepalive_tick() no-op (BAT-007, BAT-047/REQ-092) │ │ ├── messaging_handlers.py # intent/snapshot/pause RPC handlers (+update_thresholds REQ-040) + slot event payload builder (BAT-007/BAT-008/BAT-011) │ │ ├── portfolio.py # PortfolioHubReadPort + SnapshotStore + PauseStore + envelope builder (BAT-008) │ │ ├── reconciliation.py # reconcile_positions + reconcile_core_holdings(REQ-062) (BAT-008/029) @@ -86,12 +86,12 @@ │ ├── adapters/ # 인프라 어댑터 (BAT-002, BAT-004, BAT-006, BAT-007, BAT-008) │ │ ├── broker/ # paper.py(결정적 async 더블 + 현재가 stub) · kis.py(KIS Open API + 잔고조회 + 현재가 polling) (BAT-004/BAT-008/BAT-011) │ │ ├── ledger/ # in_memory.py(stub) · portfolio_hub.py(HTTP 어댑터) (BAT-006) -│ │ ├── messaging/ # topology.py(동적 큐명 빌더 + trex.direct/trex.broadcast exchange, BAT-013) · in_memory.py(stub) · rabbitmq.py(pika 동기, declare_topology(instance_key)) (BAT-007/BAT-013) +│ │ ├── messaging/ # topology.py(동적 큐명 빌더 + trex.direct/trex.broadcast exchange, BAT-013) · in_memory.py(stub, keepalive_tick no-op) · rabbitmq.py(pika 동기, declare_topology(instance_key), publish_event 재시도+강제 재연결, keepalive_tick, _publish_dlq ERROR 로그) (BAT-007/BAT-013/BAT-047) │ │ ├── portfolio/ # hub_read.py(HTTP) · stub_read.py(stub) · pause_store.py(Postgres) · killswitch_store.py(day-scoped, BAT-012) (BAT-008) │ │ └── persistence/ # postgresql.py(+PostgreSQLLedgerOutbox) · ops_checks.py(ledger-gap, transition_log 기반) · in_memory_intent_store.py · migrate.py · sql/(001~009, 010_literal_inactive_status.sql, 011_drop_pool_model_columns.sql) │ └── runtime/ # 데몬 (BAT-002, BAT-004, BAT-006, BAT-007, BAT-008) │ ├── config.py # DaemonConfig(+instance_key, max_active_slots) + KISConfig + PortfolioHubConfig + RabbitMQConfig + SnapshotScheduleConfig + SafetySellConfig + DaemonInstanceConfig.from_toml/from_config (env + per-instance TOML, BAT-013/028/030) -│ ├── daemon.py # Daemon class: `--config` TOML 진입점(BAT-013) + setup(migrate/reconcile/declare-topology(instance_key)) + 7 worker threads(instance_key 큐 구독) + WatchdogStore + KillSwitchService + graceful shutdown +│ ├── daemon.py # Daemon class: `--config` TOML 진입점(BAT-013) + setup(migrate/reconcile/declare-topology(instance_key)) + 7 worker threads(instance_key 큐 구독) + WatchdogStore + KillSwitchService + graceful shutdown. _execution_loop 각 틱 시작 시 event_publisher.keepalive_tick() 호출(REQ-092). _make_event_emitter 발행 최종 실패 시 ERROR 로그(REQ-093). │ ├── reset.py # trex-core-reset: slot-core 테이블 wipe + 자산 중심 재초기화 (REQ-064, BAT-030) │ ├── logging_setup.py # 구조화 로깅(ISO8601 KST·레벨·Loki JSON·correlation id) + configure_logging (BAT-045 / REQ-084, dcy-029) │ └── cli.py # trex-core CLI: pause/resume/paused + universe/slots/enter/exit/cancel + report(REQ-088) (BAT-008/020/046 / REQ-032) @@ -100,7 +100,8 @@ ├── domain/ # test_domain_import_sweep (NFR-1 가드, BAT-006/BAT-007) ├── application/ # test_intent_concurrency · test_execution · test_execution_ledger · test_ledger_outbox · test_messaging_intent_consumer · test_messaging_snapshot_consumer · test_messaging_event_publisher ├── runtime/ # test_config_dotenv · test_daemon_no_http_surface (BAT-007 HTTP 표면 0 회귀 가드) - └── adapters/{broker,ledger,messaging,persistence}/ # paper/KIS · in_memory/portfolio-hub ledger · stub/RabbitMQ transport (게이트) · PG repo/migration/outbox + ├── adapters/{broker,ledger,messaging,persistence}/ # paper/KIS · in_memory/portfolio-hub ledger · stub/RabbitMQ transport (게이트) · PG repo/migration/outbox + │ └── messaging/test_publish_reliability.py # BAT-047 신규 — publish 재시도/keepalive/DLQ 가시성 단위 테스트 (mock 기반, 17건) ``` > 참고: domain(BAT-001)은 무의존 코어. application/adapters/runtime 은 BAT-002 에서 추가, BAT-004 에서 broker/execution 레이어 확장, BAT-006 에서 ledger 레이어 추가, BAT-007 에서 messaging 레이어(application MessagingTransport + adapters/messaging + sync+thread 데몬) 추가와 HTTP 어댑터 제거. 외부 의존(psycopg/httpx/redis/pika)은 adapters/runtime 에만 격리(NFR-1, import 스윕으로 강제). 브로커 주문은 BrokerPort 단일 세션(불변식 B), 슬롯 쓰기는 SlotService 단일 writer(불변식 A), ledger 기록은 집행 경로 단일 주체(NFR-3), 메시지 transport 는 MessagingTransport 단일 경계. diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index 0ac9169..c4bc726 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -4,7 +4,7 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-06-10 (BAT-031 — dcy-021 포지션 사이징) +- Last Updated (KST): 2026-06-19 (BAT-047 — dcy-031 이벤트 발행 연결 신뢰성) - Source Discovery / Batch: bootstrap-baseline (pre-Discovery); 데몬/HTTP 흐름 = dcy-002 / BAT-002 - Primary Runtime: daemon (HTTP API) + library (slot 도메인) @@ -30,12 +30,12 @@ - SlotRepository (Protocol) :: 저장소 구현체 주입 경계 - MessagingTransport (Protocol) :: 메시지 transport 추상화 (RabbitMQ / InMemory stub) -## Flow: `데몬 lifecycle` (BAT-002, BAT-007) +## Flow: `데몬 lifecycle` (BAT-002, BAT-007, BAT-047) 1. `trex-core-daemon --config .toml` 기동 → `DaemonConfig.from_config(toml)` = `from_env()`(TREX_POSTGRESQL_DSN/`TREX_RABBITMQ_URL` 등) + TOML(`instance_key`, `portfolio_hub_token`, `trading_timezone`, `max_active_slots`) 병합. `--config` 필수(없으면 오류 종료, BAT-013/REQ-048). `instance_key` 로 큐명·라우팅·DB 행 스코프 분리(REQ-056). 2. setup: `apply_migrations(dsn)`(멱등; 010 inactive CHECK·011 컬럼 drop 포함) → `execution.reconcile()`(REQ-014) → 이벤트 publisher 의 `declare_topology()`. (일일 preallocation 은 BAT-030 에서 폐기 — 슬롯 생성은 universe reconcile 이 담당.) 3. start: 7개 백그라운드 스레드 동시 기동 - - `trex-execution` :: 브로커 dispatch/poll 루프 (REQ-013/014; 매도 수량 브로커 정본 REQ-062) + - `trex-execution` :: 브로커 dispatch/poll 루프 (REQ-013/014; 매도 수량 브로커 정본 REQ-062). **각 틱 시작 시 `event_publisher.keepalive_tick()` 호출** (REQ-092 — pika 스레드 안전 규칙: 발행 연결 소유 스레드에서만 I/O). 틱 간격(기본 1s) < heartbeat/2(15s) → 서버 측 idle 종료 예방. - `trex-ledger` :: ledger outbox 드레인 (REQ-019) - `trex-intent-rpc` :: `trex.intent.req` 소비자 (prefetch=1, FIFO, dedup via intent_log; `update_thresholds` 액션 포함 REQ-040) - `trex-slots-rpc` :: `trex.slots.req` 소비자 (spy=false 시 holding 슬롯 워치독 `last_seen_at` 갱신 REQ-034; `safety_sell_enabled=false` 시 `trex-safety-sell` 미기동 BAT-036) @@ -62,15 +62,25 @@ 6. ack-after-commit 시퀀스: durable commit 이 ack 보다 먼저이므로 ack 전 크래시 시 재전달돼도 (3) 의 dedup 으로 재실행되지 않는다. 7. 체결/종료는 응답이 아니라 후속 이벤트(아래)로 전달 — admission 동기 + fill 비동기 2단계 시맨틱. -## Flow: `슬롯 이벤트 발행/구독` (RabbitMQ `trex.events`, BAT-007) +## Flow: `슬롯 이벤트 발행/구독` (RabbitMQ `trex.events`, BAT-007, BAT-047) 1. 체결: execution 루프가 `pending_buy→holding`(또는 `pending_sell→closed`)을 commit + ledger outbox enqueue (REQ-018/019). -2. 같은 스레드에서 `event_publisher.publish_event(rk, payload)` 호출. +2. 같은 스레드에서 `event_publisher.publish_event(rk, payload)` 호출 (REQ-091 재시도 정책 포함, BAT-047). - routing key: `{event_type}.{strategy_id}.{symbol}` (예: `slot_filled_buy.A.005930`). - payload: 현재 상태 + `sequence_at`(슬롯 `updated_at` iso) 운반 → 수신자는 last-wins 적용 가능 (순서 보증 불요). - - publisher confirms 활성, 발행 실패는 로그 + DLQ 라우팅(`trex.events.dlq`). publish 실패가 슬롯 전이를 롤백하지 않는다. -3. 전략별 durable queue(`trex.events.strategy.{strategy_id}`)는 `*.{strategy_id}.#` 패턴으로 바인딩 → broker 단 필터로 자기 자산 이벤트만 수신. -4. 전략 down 중 적체된 메시지는 catch-up 으로 복구; TTL(1h) / max-length(10k) 초과는 `drop-head` + 후속 스냅샷 RPC reconcile. + - publisher confirms 활성. `is_open=True`를 맹신하지 않음 — 발행 실패 시(half-open 포함) 강제 재연결(`close`+`_connect(timeout=2s)`) 후 1회 재시도. 재시도도 실패하면 `TransportUnavailableError` 전파. + - unroutable(구독자 없음): `UnroutableError` → `_publish_dlq`로 `trex.events.dlq`(durable)에 보존. DLQ 유입 시 ERROR 로그 발행(REQ-093). + - publish 최종 실패는 ERROR 로그(routing key + slot_id + strategy_id 포함)로 경보 — 슬롯 전이 롤백 없음(NFR-3). +3. idle keepalive: `trex-execution` 스레드가 각 틱 시작 시 `event_publisher.keepalive_tick()`(`process_data_events(time_limit=0)`)을 호출 → heartbeat 주기 내 I/O 보장 → 서버 측 idle 종료 예방 (1차 방어선, REQ-092). 발행 재시도는 2차 복구 방어선(REQ-091). +4. 전략별 durable queue(`trex.events.strategy.{strategy_id}`)는 `*.{strategy_id}.#` 패턴으로 바인딩 → broker 단 필터로 자기 자산 이벤트만 수신. +5. 전략 down 중 적체된 메시지는 catch-up 으로 복구; TTL(1h) / max-length(10k) 초과는 `drop-head` + 후속 스냅샷 RPC reconcile. + +### DLQ 관측 경로 (`trex.events.dlq`, REQ-093) + +- unroutable 이벤트 → `_publish_dlq(reason="unroutable")` → ERROR 로그 + `trex.events.dlq`(durable) 유입. +- 발행 최종 실패 → `_make_event_emitter` ERROR 로그(경보). DLQ 유입 없음(연결 자체가 없는 경우). +- `_publish_dlq` 재발행 자체 실패 시 → ERROR 로그(`DLQ publish failed rk=...`). +- 운영자 확인: 로그 `level=ERROR message=~"DLQ"` 또는 RabbitMQ 관리 UI queue depth `trex.events.dlq` 조회. ## Flow: `일일 스냅샷 refresh` (BAT-008 / REQ-030) diff --git a/docs/srs/Documentation/req-094_event-publisher-reliability-baseline-alignment.md b/docs/srs/Documentation/req-094_event-publisher-reliability-baseline-alignment.md index c178e29..3c55626 100644 --- a/docs/srs/Documentation/req-094_event-publisher-reliability-baseline-alignment.md +++ b/docs/srs/Documentation/req-094_event-publisher-reliability-baseline-alignment.md @@ -1,6 +1,6 @@ # req-094: 이벤트 발행 신뢰성 baseline 문서 정합 (runtime-flows / project-structure / operations 갱신) -- Status: Approved +- Status: Implemented - Type: Documentation - Priority: Medium - Owner: project maintainers diff --git a/docs/srs/Exception/req-091_publish-retry-and-force-reconnect.md b/docs/srs/Exception/req-091_publish-retry-and-force-reconnect.md index ca5c6ca..2a4ced1 100644 --- a/docs/srs/Exception/req-091_publish-retry-and-force-reconnect.md +++ b/docs/srs/Exception/req-091_publish-retry-and-force-reconnect.md @@ -1,6 +1,6 @@ # req-091: 이벤트 발행 재시도 + 강제 재연결 (idle 후 첫 발행 무유실) -- Status: Approved +- Status: Implemented - Type: Exception - Priority: High - Owner: project maintainers diff --git a/docs/srs/Exception/req-092_publish-idle-keepalive.md b/docs/srs/Exception/req-092_publish-idle-keepalive.md index ae15fd3..fe6daeb 100644 --- a/docs/srs/Exception/req-092_publish-idle-keepalive.md +++ b/docs/srs/Exception/req-092_publish-idle-keepalive.md @@ -1,6 +1,6 @@ # req-092: 발행 전용 연결 idle keepalive (서버 측 heartbeat 사망 예방) -- Status: Approved +- Status: Implemented - Type: Exception - Priority: High - Owner: project maintainers diff --git a/docs/srs/Exception/req-093_dlq-visibility-and-emit-error-alert.md b/docs/srs/Exception/req-093_dlq-visibility-and-emit-error-alert.md index 46330ec..5048d71 100644 --- a/docs/srs/Exception/req-093_dlq-visibility-and-emit-error-alert.md +++ b/docs/srs/Exception/req-093_dlq-visibility-and-emit-error-alert.md @@ -1,6 +1,6 @@ # req-093: DLQ 가시성/경보 + emit 실패 경보 (trex.events.dlq 관측성) -- Status: Approved +- Status: Implemented - Type: Exception - Priority: High - Owner: project maintainers diff --git a/docs/srs/index.md b/docs/srs/index.md index 24d995e..d6dede7 100644 --- a/docs/srs/index.md +++ b/docs/srs/index.md @@ -171,13 +171,14 @@ | REQ-088 | 운영 조회 CLI — 보유·오늘 체결·실현 PnL·대기 주문 | Interface | Implemented | High | project maintainers | [req-088](./Interface/req-088_operator-pnl-query-cli.md) | | REQ-089 | 운영자 런북 — docs/operations.md 신설 | Documentation | Implemented | Medium | project maintainers | [req-089](./Documentation/req-089_operator-runbook.md) | | REQ-090 | 관측성 baseline 문서 정합 — 로깅/관측성·PnL 조회 흐름 | Documentation | Implemented | Medium | project maintainers | [req-090](./Documentation/req-090_observability-baseline-alignment.md) | -| REQ-091 | 이벤트 발행 재시도 + 강제 재연결 (idle 후 첫 발행 무유실) | Exception | Approved | High | project maintainers | [req-091](./Exception/req-091_publish-retry-and-force-reconnect.md) | -| REQ-092 | 발행 전용 연결 idle keepalive (서버 측 heartbeat 사망 예방) | Exception | Approved | High | project maintainers | [req-092](./Exception/req-092_publish-idle-keepalive.md) | -| REQ-093 | DLQ 가시성/경보 + emit 실패 경보 (trex.events.dlq 관측성) | Exception | Approved | High | project maintainers | [req-093](./Exception/req-093_dlq-visibility-and-emit-error-alert.md) | -| REQ-094 | 이벤트 발행 신뢰성 baseline 문서 정합 (runtime-flows / project-structure / operations 갱신) | Documentation | Approved | Medium | project maintainers | [req-094](./Documentation/req-094_event-publisher-reliability-baseline-alignment.md) | +| REQ-091 | 이벤트 발행 재시도 + 강제 재연결 (idle 후 첫 발행 무유실) | Exception | Implemented | High | project maintainers | [req-091](./Exception/req-091_publish-retry-and-force-reconnect.md) | +| REQ-092 | 발행 전용 연결 idle keepalive (서버 측 heartbeat 사망 예방) | Exception | Implemented | High | project maintainers | [req-092](./Exception/req-092_publish-idle-keepalive.md) | +| REQ-093 | DLQ 가시성/경보 + emit 실패 경보 (trex.events.dlq 관측성) | Exception | Implemented | High | project maintainers | [req-093](./Exception/req-093_dlq-visibility-and-emit-error-alert.md) | +| REQ-094 | 이벤트 발행 신뢰성 baseline 문서 정합 (runtime-flows / project-structure / operations 갱신) | Documentation | Implemented | Medium | project maintainers | [req-094](./Documentation/req-094_event-publisher-reliability-baseline-alignment.md) | ## Recent Change Log Summary +- 2026-06-19 :: REQ-091·092·093·094 Approved → **Implemented** (confirm-req-implemented, BAT-047 release-candidate, dcy-031/OPS-005): 발행 재시도+강제재연결(091)·idle keepalive(092, execution-loop 단일스레드 틱)·DLQ 가시성+emit ERROR 경보(093, 구조화 로그 dcy-029 재사용)·baseline 정합(094) 구현·검증 완료. 독립 opus QC=PASS-WITH-NOTES→시정 2건(ruff F401·AC-3 프로덕션 경로) 완료, 팀장 재확인 **554 passed/47 skipped/회귀 0/ruff clean**. NFR-1 wire 무회귀(interface-contract diff 0). 수용 한계: half-open/broker재시작/idle=mock unit만(실 broker 통합 skip→운영/CI 위임), NFR-2 ≤5초는 재연결+재시도 구간만 엄밀(1차 socket_timeout 5s 최악 초과 가능). OQ-1/3 design Answered. 다음=draft-release(배포=trex-core 재배포→사장님 결재, tsl 라이브 선행). - 2026-06-19 :: REQ-091·092·093·094 Proposed → **Approved** (confirm-req, 팀장): 필수 필드·구체 AC 충족, source dcy-031(confirmed, Open OQ 0) 정합, blocker 없음. 팀장 판정 — FR-4의 REQ-093 흡수 동의, OQ-1/OQ-3 design 인계 확정, NFR-2 타임아웃 수치는 draft-batch-design 확정(비차단). 다음=suggest-batch-reqs + draft-batch(REQ-091~094 단일 batch 편성 예정, 고위험 opus QC). - 2026-06-19 :: REQ-091·092·093·094 신규 생성(Proposed, draft-req dcy-031): 이벤트 발행 연결 신뢰성 — REQ-091 발행 재시도+강제 재연결(FR-1/B, S-1·S-2, NFR-1~3 흡수), REQ-092 idle keepalive(FR-2/A, OQ-1 design 인계), REQ-093 DLQ 가시성+emit 경보(FR-3/D + FR-4, OQ-3 design 인계, S-3), REQ-094 baseline 문서 정합(runtime-flows/project-structure/operations, docs전용). NFR-2·NFR-3은 REQ-091~093 AC에 흡수. 고위험(M1 라이브) batch — opus QC 강제(CONFIRM①). diff --git a/tests/adapters/messaging/test_publish_reliability.py b/tests/adapters/messaging/test_publish_reliability.py new file mode 100644 index 0000000..294198d --- /dev/null +++ b/tests/adapters/messaging/test_publish_reliability.py @@ -0,0 +1,455 @@ +"""BAT-047 — Event publisher connection reliability tests (REQ-091/092/093). + +All tests run without a live RabbitMQ broker using unittest.mock. The +RabbitMQTransport is exercised through its public interface; pika internals +are replaced by lightweight mocks so the retry / keepalive / DLQ logic can +be unit-tested deterministically. + +AC coverage: + REQ-091 AC-1 : half-open → reconnect + retry → success (no error raised) + REQ-091 AC-2 : half-open + retry also fails → TransportUnavailableError + REQ-091 AC-3 : timeout budget — _force_reconnect uses connect_timeout=2.0 + REQ-091 AC-4 : normal publish (conn alive) — no retry overhead + REQ-091 AC-5 : consume/RPC paths untouched by this change (interface check) + REQ-091 AC-6 : broker restart scenario — reconnect + retry succeeds + REQ-092 AC-1 : keepalive_tick calls process_data_events on live conn + REQ-092 AC-2 : keepalive_tick executes only on owning object (no lock escape) + REQ-092 AC-3 : simultaneous publish + keepalive — no crash (single-thread) + REQ-092 AC-4 : keepalive non-blocking (time_limit=0 passed to pika) + REQ-092 AC-5 : consume/RPC paths unaffected (keepalive not called there) + REQ-093 AC-1 : unroutable → _publish_dlq called + ERROR log emitted + REQ-093 AC-3 : emit ERROR on final publish failure (daemon layer) + REQ-093 AC-4 : _publish_dlq itself fails → ERROR log, no silent loss + REQ-093 AC-6 : consume/RPC paths unaffected (DLQ change local to publish) +""" +from __future__ import annotations + +import logging +from unittest.mock import MagicMock, patch + +import pika.exceptions +import pytest + +from trex_core.adapters.messaging.rabbitmq import RabbitMQTransport +from trex_core.application.messaging import TransportUnavailableError + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_transport() -> RabbitMQTransport: + """Return a RabbitMQTransport instance without connecting to a real broker.""" + return RabbitMQTransport("amqp://localhost/", timeout=5.0) + + +def _inject_connected(t: RabbitMQTransport, conn=None, channel=None): + """Put the transport into CONNECTED state with mock pika objects.""" + from trex_core.application.messaging import TransportHealth + if conn is None: + conn = MagicMock() + conn.is_open = True + if channel is None: + channel = MagicMock() + channel.is_open = True + t._conn = conn + t._channel = channel + t._health = TransportHealth.CONNECTED + return conn, channel + + +# --------------------------------------------------------------------------- +# REQ-091 AC-1: half-open → reconnect + retry → success +# --------------------------------------------------------------------------- + +class TestPublishRetryOnHalfOpen: + """REQ-091 AC-1: publish succeeds after forced reconnect.""" + + def test_half_open_first_attempt_fails_retry_succeeds(self, caplog): + """Simulates a half-open connection: is_open=True but basic_publish raises + AMQPConnectionError on first call. The transport must reconnect and retry, + and publish_event must NOT raise. + """ + t = _make_transport() + mock_conn, mock_ch = _inject_connected(t) + + call_count = {"n": 0} + + def fake_basic_publish(**kwargs): + call_count["n"] += 1 + if call_count["n"] == 1: + raise pika.exceptions.AMQPConnectionError("connection reset") + # Second call (after reconnect) succeeds silently. + + mock_ch.basic_publish.side_effect = fake_basic_publish + + reconnect_called = {"yes": False} + + def fake_force_reconnect(): + reconnect_called["yes"] = True + # After reconnect, the channel is fresh and functional. + new_ch = MagicMock() + new_ch.is_open = True + new_ch.basic_publish.return_value = None # success + t._channel = new_ch + from trex_core.application.messaging import TransportHealth + t._health = TransportHealth.CONNECTED + + with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect): + # Must not raise. + t.publish_event("slot_filled_buy.A.005930", {"slot": "s1"}) + + assert reconnect_called["yes"], "_force_reconnect must be called on first failure" + + +# --------------------------------------------------------------------------- +# REQ-091 AC-2: retry also fails → TransportUnavailableError raised +# --------------------------------------------------------------------------- + +class TestPublishRetryExhausted: + """REQ-091 AC-2: when reconnect+retry also fails, error is propagated.""" + + def test_retry_exhausted_raises_transport_unavailable(self): + t = _make_transport() + mock_conn, mock_ch = _inject_connected(t) + + mock_ch.basic_publish.side_effect = pika.exceptions.AMQPConnectionError("gone") + + def fake_force_reconnect(): + from trex_core.application.messaging import TransportHealth + new_ch = MagicMock() + new_ch.is_open = True + new_ch.basic_publish.side_effect = pika.exceptions.AMQPConnectionError( + "still gone" + ) + t._channel = new_ch + t._health = TransportHealth.CONNECTED + + with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect): + with pytest.raises(TransportUnavailableError) as exc_info: + t.publish_event("slot_filled_buy.A.005930", {"slot": "s1"}) + + assert "reconnect+retry" in str(exc_info.value) + + +# --------------------------------------------------------------------------- +# REQ-091 AC-3: timeout budget — _force_reconnect uses connect_timeout=2.0 +# --------------------------------------------------------------------------- + +class TestForceReconnectTimeout: + """REQ-091 AC-3 / NFR-2: _force_reconnect passes connect_timeout=2.0.""" + + def test_force_reconnect_uses_2s_timeout(self): + t = _make_transport() + _inject_connected(t) + + connect_calls: list = [] + + def fake_connect(connect_timeout=None): + connect_calls.append(connect_timeout) + from trex_core.application.messaging import TransportHealth + new_conn = MagicMock() + new_conn.is_open = True + new_ch = MagicMock() + new_ch.is_open = True + t._conn = new_conn + t._channel = new_ch + t._health = TransportHealth.CONNECTED + + with patch.object(t, "_connect", side_effect=fake_connect): + t._force_reconnect() + + assert len(connect_calls) == 1 + assert connect_calls[0] == 2.0, ( + f"_force_reconnect must pass connect_timeout=2.0, got {connect_calls[0]}" + ) + + +# --------------------------------------------------------------------------- +# REQ-091 AC-4: normal publish (conn alive) — no retry path triggered +# --------------------------------------------------------------------------- + +class TestNormalPublishNoRetry: + """REQ-091 AC-4: successful first-attempt publish does not invoke retry.""" + + def test_normal_publish_does_not_call_force_reconnect(self): + t = _make_transport() + _inject_connected(t) + t._channel.basic_publish.return_value = None # success + + with patch.object(t, "_force_reconnect") as mock_fr: + t.publish_event("slot_filled_buy.A.005930", {"ok": True}) + + mock_fr.assert_not_called() + + +# --------------------------------------------------------------------------- +# REQ-091 AC-6: broker restart — force reconnect + retry succeeds +# --------------------------------------------------------------------------- + +class TestBrokerRestartScenario: + """REQ-091 AC-6: after broker restart the transport reconnects and retries.""" + + def test_broker_restart_publish_recovers(self): + t = _make_transport() + _inject_connected(t) + + # Simulate broker restart: first publish raises OSError (TCP reset). + t._channel.basic_publish.side_effect = OSError("connection refused") + + recovered = {"yes": False} + + def fake_force_reconnect(): + from trex_core.application.messaging import TransportHealth + new_ch = MagicMock() + new_ch.is_open = True + new_ch.basic_publish.return_value = None + t._channel = new_ch + t._health = TransportHealth.CONNECTED + recovered["yes"] = True + + with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect): + # publish_event must succeed without raising. + t.publish_event("slot_filled_buy.A.005930", {"ev": "restart"}) + + assert recovered["yes"] + + +# --------------------------------------------------------------------------- +# REQ-092 AC-1: keepalive_tick calls process_data_events on live conn +# --------------------------------------------------------------------------- + +class TestKeepaliveTick: + """REQ-092 AC-1: keepalive_tick delegates to conn.process_data_events.""" + + def test_keepalive_tick_calls_process_data_events_on_live_connection(self): + t = _make_transport() + mock_conn, _ = _inject_connected(t) + mock_conn.is_open = True + + t.keepalive_tick() + + mock_conn.process_data_events.assert_called_once_with(time_limit=0) + + def test_keepalive_tick_noop_when_disconnected(self): + """REQ-092 AC-1 variant: no-op if connection is None (shutdown path).""" + t = _make_transport() + # _conn is None by default — keepalive_tick must not raise. + t.keepalive_tick() # should return silently + + def test_keepalive_tick_noop_when_conn_closed(self): + """keepalive_tick skips when is_open=False.""" + t = _make_transport() + mock_conn = MagicMock() + mock_conn.is_open = False + t._conn = mock_conn + from trex_core.application.messaging import TransportHealth + t._health = TransportHealth.DISCONNECTED + + t.keepalive_tick() + + mock_conn.process_data_events.assert_not_called() + + +# --------------------------------------------------------------------------- +# REQ-092 AC-4: keepalive non-blocking — time_limit=0 +# --------------------------------------------------------------------------- + +class TestKeepaliveTimeLimitZero: + """REQ-092 AC-4: process_data_events is called with time_limit=0.""" + + def test_keepalive_uses_time_limit_zero(self): + t = _make_transport() + mock_conn, _ = _inject_connected(t) + mock_conn.is_open = True + + t.keepalive_tick() + + args, kwargs = mock_conn.process_data_events.call_args + assert kwargs.get("time_limit", args[0] if args else None) == 0 + + +# --------------------------------------------------------------------------- +# REQ-092 AC-3: publish + keepalive in same thread — no crash +# --------------------------------------------------------------------------- + +class TestKeepaliveAndPublishSameThread: + """REQ-092 AC-3: calling keepalive then publish in the same thread is safe.""" + + def test_keepalive_then_publish_sequential_no_error(self): + t = _make_transport() + mock_conn, _ = _inject_connected(t) + mock_conn.is_open = True + t._channel.basic_publish.return_value = None + + # Sequential in same thread — simulates execution-loop tick. + t.keepalive_tick() + t.publish_event("slot_filled_buy.A.005930", {"seq": 1}) + + mock_conn.process_data_events.assert_called_once_with(time_limit=0) + t._channel.basic_publish.assert_called_once() + + +# --------------------------------------------------------------------------- +# REQ-092 AC-5: consume/RPC paths are unaffected by keepalive +# --------------------------------------------------------------------------- + +class TestKeepaliveDoesNotAffectConsumePaths: + """REQ-092 AC-5: keepalive is specific to publish path; consume API unchanged.""" + + def test_keepalive_tick_not_present_on_consume_rpc_forever_signature(self): + """The consume_rpc_forever signature is unchanged.""" + import inspect + sig = inspect.signature(RabbitMQTransport.consume_rpc_forever) + params = list(sig.parameters) + assert "queue" in params + assert "handler" in params + assert "stop" in params + + +# --------------------------------------------------------------------------- +# REQ-093 AC-1: unroutable → _publish_dlq called + ERROR log emitted +# --------------------------------------------------------------------------- + +class TestDlqVisibilityOnUnroutable: + """REQ-093 AC-1: unroutable event triggers _publish_dlq + ERROR log.""" + + def test_unroutable_calls_publish_dlq_and_logs_error(self, caplog): + t = _make_transport() + _inject_connected(t) + + t._channel.basic_publish.side_effect = pika.exceptions.UnroutableError([]) + + dlq_calls: list = [] + + def fake_dlq(rk, payload, *, reason): + dlq_calls.append((rk, reason)) + + with patch.object(t, "_publish_dlq", side_effect=fake_dlq): + with caplog.at_level(logging.DEBUG): + # Must not raise — unroutable is handled via DLQ. + t.publish_event("slot_filled_buy.A.005930", {"x": 1}) + + assert len(dlq_calls) == 1 + assert dlq_calls[0][1] == "unroutable" + + def test_publish_dlq_logs_error_on_entry(self, caplog): + """REQ-093 AC-1: _publish_dlq itself logs ERROR when DLQ entry is created.""" + t = _make_transport() + _inject_connected(t) + # Make DLQ publish succeed (no-op). + t._channel.basic_publish.return_value = None + + with caplog.at_level(logging.ERROR, logger="trex_core.adapters.messaging.rabbitmq"): + t._publish_dlq("slot_filled_buy.A.005930", {"x": 1}, reason="unroutable") + + errors = [r for r in caplog.records if r.levelno >= logging.ERROR] + assert any("DLQ" in r.message for r in errors), ( + "Expected ERROR log with 'DLQ' when _publish_dlq is called" + ) + + +# --------------------------------------------------------------------------- +# REQ-093 AC-4: _publish_dlq itself fails → ERROR log, no silent loss +# --------------------------------------------------------------------------- + +class TestDlqPublishFailureLogsError: + """REQ-093 AC-4: when DLQ publish itself raises, ERROR is logged.""" + + def test_dlq_publish_failure_logs_error_not_warning(self, caplog): + t = _make_transport() + _inject_connected(t) + + # First basic_publish for DLQ raises (after _ensure_connected succeeds). + t._channel.basic_publish.side_effect = Exception("DLQ broker down") + + with caplog.at_level(logging.ERROR, logger="trex_core.adapters.messaging.rabbitmq"): + # _publish_dlq must swallow the exception but log ERROR. + t._publish_dlq("slot_filled_buy.A.005930", {"x": 1}, reason="unroutable") + + error_records = [r for r in caplog.records if r.levelno >= logging.ERROR] + assert len(error_records) >= 1 + # The DLQ publish failure must be ERROR, not WARNING. + dlq_fail_records = [r for r in error_records if "DLQ" in r.message] + assert dlq_fail_records, "Expected ERROR log mentioning 'DLQ' on publish failure" + + +# --------------------------------------------------------------------------- +# REQ-093 AC-3: emit ERROR on final publish failure (daemon layer) +# --------------------------------------------------------------------------- + +class TestEmitErrorOnFinalPublishFailure: + """REQ-093 AC-3: _make_event_emitter logs ERROR (not WARNING) on failure.""" + + def test_emit_logs_error_on_transport_unavailable(self, caplog): + """The emitter created by _make_event_emitter must log ERROR when + publish_event raises TransportUnavailableError after retry. + + Uses the production Daemon._make_event_emitter() directly (via a + lightweight stub instance) so that any change to the production error + path is caught as a regression. + """ + from trex_core.runtime.daemon import Daemon + from trex_core.application.messaging import TransportUnavailableError + + # Construct a minimal Daemon stub without triggering __init__ (which + # requires DB / config / broker connections). Only event_publisher is + # accessed by _make_event_emitter, so that is the only attribute we set. + daemon_stub = object.__new__(Daemon) + fake_publisher = MagicMock() + fake_publisher.publish_event.side_effect = TransportUnavailableError("no broker") + daemon_stub.event_publisher = fake_publisher + + # Obtain the emitter from the production code path. + emit = daemon_stub._make_event_emitter() + + # Minimal slot stub. + slot = MagicMock() + slot.strategy_id = "A" + slot.symbol = "005930" + slot.slot_id = "slot-1" + slot.status = MagicMock() + slot.status.value = "holding" + slot.updated_at = None + slot.portfolio_id = "PF-1" + slot.entry_price = None + slot.qty = 10 + slot.instance_key = "default" + slot.actor = None + slot.reason = None + + with caplog.at_level(logging.ERROR, logger="trex_core.runtime.daemon"): + emit("slot_filled_buy", slot) + + error_records = [ + r for r in caplog.records + if r.levelno >= logging.ERROR and "event publish failed" in r.message + ] + assert error_records, ( + "Expected ERROR log 'event publish failed' from Daemon._make_event_emitter" + ) + # Must NOT be mere WARNING. + warn_only = [ + r for r in caplog.records + if r.levelno == logging.WARNING and "event publish failed" in r.message + ] + assert not warn_only, "publish failure must not be WARNING; must be ERROR" + + +# --------------------------------------------------------------------------- +# REQ-093 AC-6: consume/RPC paths unaffected +# --------------------------------------------------------------------------- + +class TestConsumePaths: + """REQ-093 AC-6 / REQ-091 AC-5: consume/RPC interface signature unchanged.""" + + def test_rabbitmq_transport_has_consume_methods(self): + t = _make_transport() + assert callable(getattr(t, "consume_rpc_forever", None)) + assert callable(getattr(t, "consume_events_forever", None)) + assert callable(getattr(t, "call_rpc", None)) + assert callable(getattr(t, "declare_topology", None)) + + def test_keepalive_tick_exists_as_callable(self): + t = _make_transport() + assert callable(getattr(t, "keepalive_tick", None)) diff --git a/trex_core/adapters/messaging/in_memory.py b/trex_core/adapters/messaging/in_memory.py index a6440fa..80b6eac 100644 --- a/trex_core/adapters/messaging/in_memory.py +++ b/trex_core/adapters/messaging/in_memory.py @@ -144,6 +144,9 @@ class InMemoryTransport: if entry in self._event_subscribers: self._event_subscribers.remove(entry) + def keepalive_tick(self) -> None: + """No-op for in-memory transport (REQ-092 — RabbitMQ only).""" + def close(self) -> None: with self._lock: self._available = False diff --git a/trex_core/adapters/messaging/rabbitmq.py b/trex_core/adapters/messaging/rabbitmq.py index 1626e51..221fa0f 100644 --- a/trex_core/adapters/messaging/rabbitmq.py +++ b/trex_core/adapters/messaging/rabbitmq.py @@ -76,10 +76,11 @@ class RabbitMQTransport: self._health = TransportHealth.DISCONNECTED # ------------------------------------------------------------ lifecycle - def _connect(self) -> None: + def _connect(self, connect_timeout: Optional[float] = None) -> None: params = pika.URLParameters(self._url) - params.socket_timeout = self._timeout - params.blocked_connection_timeout = self._timeout + t = connect_timeout if connect_timeout is not None else self._timeout + params.socket_timeout = t + params.blocked_connection_timeout = t params.heartbeat = 30 with self._lock: self._health = TransportHealth.RECONNECTING @@ -102,6 +103,36 @@ class RabbitMQTransport: if not ok: self._connect() + def _force_reconnect(self) -> None: + """Force-close current connection and reconnect (REQ-091). + + Uses a shorter socket_timeout (2 s) so a single publish attempt stays + well within the 5 s execution-loop block budget (NFR-2). + """ + self.close() + self._connect(connect_timeout=2.0) + + def keepalive_tick(self) -> None: + """Process pending I/O frames on the publish connection (REQ-092). + + MUST be called from the thread that owns this transport instance + (pika BlockingConnection is NOT thread-safe). The execution-loop + calls this once per tick (interval < heartbeat/2) so the broker's + idle-connection reaper never fires. + + Uses time_limit=0 to be non-blocking — only drains already-pending + frames from the socket buffer. + """ + with self._lock: + conn = self._conn + if conn is None or not conn.is_open: + return + try: + conn.process_data_events(time_limit=0) + except Exception: # noqa: BLE001 — keepalive is best-effort + _log.debug("keepalive_tick process_data_events failed", exc_info=True) + self._mark_disconnected() + def close(self) -> None: with self._lock: ch, conn = self._channel, self._conn @@ -165,6 +196,27 @@ class RabbitMQTransport: ch.queue_bind(queue, EVENTS_EXCHANGE, routing_key=binding_pattern) # ------------------------------------------------------------ publish + def _do_basic_publish( + self, + routing_key: str, + body: bytes, + *, + mandatory: bool, + ) -> None: + """Internal: call basic_publish on current channel (must be connected).""" + ch = self._channel + assert ch is not None + ch.basic_publish( + exchange=EVENTS_EXCHANGE, + routing_key=routing_key, + body=body, + properties=pika.BasicProperties( + content_type="application/json", + delivery_mode=2, # persistent + ), + mandatory=mandatory, + ) + def publish_event( self, routing_key: str, @@ -172,31 +224,56 @@ class RabbitMQTransport: *, mandatory: bool = True, ) -> None: + """Publish to trex.events exchange with retry-on-failure (REQ-091). + + Strategy: + 1. Attempt publish (after _ensure_connected). + 2. On AMQPError/OSError (including half-open stale connections): + force-reconnect, then retry once. + 3. If retry also fails: raise TransportUnavailableError. + 4. UnroutableError (no subscribers) → DLQ, no retry needed. + + is_open=True is NOT trusted alone — a half-open connection will pass + the is_open check but fail on basic_publish. The retry path handles + this transparently (REQ-091 FR-1). + """ body = json.dumps(payload).encode("utf-8") try: self._ensure_connected() - ch = self._channel - assert ch is not None - ch.basic_publish( - exchange=EVENTS_EXCHANGE, - routing_key=routing_key, - body=body, - properties=pika.BasicProperties( - content_type="application/json", - delivery_mode=2, # persistent - ), - mandatory=mandatory, - ) + self._do_basic_publish(routing_key, body, mandatory=mandatory) except pika.exceptions.UnroutableError: # No matching queue subscribers → push to DLQ collector (REQ-023 AC-6). self._publish_dlq(routing_key, payload, reason="unroutable") - except (pika.exceptions.AMQPError, OSError) as exc: + except (pika.exceptions.AMQPError, OSError) as first_exc: + # First attempt failed (possibly half-open). Force reconnect + retry once. + _log.debug( + "publish_event first attempt failed, force-reconnecting (rk=%s): %s", + routing_key, first_exc, + ) self._mark_disconnected() - raise TransportUnavailableError( - f"publish_event failed: {exc.__class__.__name__}: {exc}" - ) from exc + try: + self._force_reconnect() + self._do_basic_publish(routing_key, body, mandatory=mandatory) + except pika.exceptions.UnroutableError: + # Reconnected but still unroutable after retry. + self._publish_dlq(routing_key, payload, reason="unroutable") + except (pika.exceptions.AMQPError, OSError) as retry_exc: + self._mark_disconnected() + raise TransportUnavailableError( + f"publish_event failed after reconnect+retry: " + f"{retry_exc.__class__.__name__}: {retry_exc}" + ) from retry_exc def _publish_dlq(self, routing_key: str, payload: dict, *, reason: str) -> None: + """Best-effort publish to DLQ (REQ-093). + + Logs ERROR on both DLQ entry (visibility) and DLQ publish failure + so the event is never silently lost from the operations perspective. + """ + _log.error( + "DLQ entry: reason=%s rk=%s payload_keys=%s", + reason, routing_key, list(payload.keys()), + ) try: self._ensure_connected() ch = self._channel @@ -213,7 +290,7 @@ class RabbitMQTransport: ), ) except Exception: # noqa: BLE001 - DLQ best-effort - _log.warning("DLQ publish failed", exc_info=True) + _log.error("DLQ publish failed (rk=%s)", routing_key, exc_info=True) # ------------------------------------------------------------ RPC client def call_rpc( diff --git a/trex_core/application/messaging.py b/trex_core/application/messaging.py index 6debcab..3da13a5 100644 --- a/trex_core/application/messaging.py +++ b/trex_core/application/messaging.py @@ -112,6 +112,15 @@ class MessagingTransport(Protocol): exchange, dispatch to handler. Blocks until `stop` is set.""" ... + def keepalive_tick(self) -> None: + """Process pending I/O frames to keep the connection alive (REQ-092). + + MUST be called from the thread that owns this transport instance. + Non-blocking (time_limit=0 equivalent). Implementations that do not + need keepalive (e.g. InMemoryTransport) should be a no-op. + """ + ... + def health(self) -> TransportHealth: """Current connection state (REQ-024).""" ... diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index e38e7bc..20c73e1 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -244,12 +244,24 @@ class Daemon: try: self.event_publisher.publish_event(rk, payload) except Exception: # noqa: BLE001 - publish failure must not roll back - _log.warning("event publish failed (rk=%s)", rk, exc_info=True) + # REQ-093: escalate to ERROR so operators can observe via log + # alert. Include routing key + slot identity for correlation. + _log.error( + "event publish failed (rk=%s slot_id=%s strategy=%s)", + rk, + getattr(slot, "slot_id", None) or getattr(slot, "id", None), + getattr(slot, "strategy_id", None), + exc_info=True, + ) return emit # ------------------------------------------------------------ workers def _execution_loop(self) -> None: while not self.stop.wait(self.config.execution_interval_sec): + # REQ-092: keepalive tick — process pending I/O frames on the + # publish connection before each execution run. Called in the + # same thread that owns event_publisher (pika thread-safety rule). + self.event_publisher.keepalive_tick() try: self.execution.run_once() except Exception: # noqa: BLE001 - loop must survive transient errors