diff --git a/.proofline/criteria/ac-0040.md b/.proofline/criteria/ac-0040.md new file mode 100644 index 0000000..aa2e4a6 --- /dev/null +++ b/.proofline/criteria/ac-0040.md @@ -0,0 +1,23 @@ +--- +id: "ac-0040" +status: active +--- + +# Ledger outbox idle poll 로그 억제와 운영 metric + +## Criterion + +Ledger outbox writer는 pending row가 없는 poll마다 instance-local process-lifetime empty poll counter를 정확히 증가시키되 empty claim INFO를 생성하지 않아야 한다. non-empty batch claim은 configured `instance_key`, non-zero `count`, `limit`, 해당 batch의 non-empty correlation ID를 repository claim 순서와 동일한 `correlation_ids` JSON string array로 INFO에 남겨야 한다. empty poll은 claim INFO 자체를 생성하지 않으므로 `correlation_ids` field도 출력하지 않는다. + +`trex.slots.req`의 `spy=true` 응답은 top-level `ledger_outbox_metrics`를 포함해야 한다. 정상 snapshot은 `available=true`, `scope=process_lifetime`, configured `instance_key`를 포함하고, `claims` object에서 `empty_polls_total`, `claimed_batches_total`, `unavailable_total`, `lost_total`, `foreign_instance_total`을, `outcomes` object에서 `completed_total`, `duplicate_completed_total`, `retryable_failure_total`, `conflict_blocked_total`, `permanent_failure_total`, `dead_lettered_total`을 각각 누락 없는 0 이상의 정수로 제공해야 한다. snapshot은 해당 daemon process의 configured instance writer만 집계하고 내부 mutable counter와 분리된 새 값이어야 한다. reader가 없거나 조회가 실패하면 `available=false`, `scope=process_lifetime`, configured `instance_key`, `reason=ledger_outbox_metrics_unavailable`만 포함한 snapshot을 반환하며 slots 조회 자체는 성공해야 한다. `spy=false` 응답에는 `ledger_outbox_metrics` key를 포함하지 않아야 한다. + +`TrexDaemon._ledger_loop()`에서 writer가 예상 밖 예외를 던지면 `reason=ledger_writer_run_once_failed`, configured `instance_key`, exception class name인 `exc_type`만 ERROR로 남기고 다음 configured poll을 계속해야 한다. empty/non-empty/failure 관측 어디에도 credential, account identifier, token, password, DSN, provider payload, raw exception message, traceback 또는 `exc_info`가 포함되어서는 안 된다. fixed polling cadence, retry scheduling, claim lease, ownership/filtering과 기존 outcome 의미는 바뀌지 않아야 한다. + +## Verification + +- 성공 조건: 기본 `ledger_interval_sec=1.0`의 24시간에 해당하는 86,400회 empty `run_once()`를 실제 sleep 없이 `instance-a`와 `instance-b` writer에 각각 실행했을 때 각 `empty_polls_total`이 정확히 86,400이고 rendered INFO의 `claim_result=empty` 및 `correlation_ids`가 0건인지 확인한다. 한 개 이상의 item을 claim하면 batch INFO에 configured instance, non-zero count, limit와 repository claim 순서 그대로인 `correlation_ids` JSON string array가 있고 `claimed_batches_total` 및 처리 outcome totals가 맞는지 확인한다. +- 운영 조회 조건: `spy=true`에서 정상 schema의 모든 key와 zero/default 정수, instance/process 격리, 매 호출 새 snapshot을 확인한다. reader absent 및 reader exception에서는 stable unavailable snapshot과 slots 응답 성공을 확인한다. `spy=false`에는 metrics key가 없음을 확인한다. +- 실패·보안 조건: sentinel secret을 포함한 writer exception을 두 번의 loop tick 중 첫 tick에 주입해 ERROR reason·instance·exception type, 두 번째 tick 실행, raw sentinel·exception message·traceback·`exc_info` 부재를 확인한다. 기존 claim-lost, foreign/unowned, retryable, conflict, permanent, dead-letter, success와 duplicate outcome의 severity·counter·상태 전이가 유지되는지 확인한다. +- 변경 없음 조건: `AC-0029`의 InMemory/PostgreSQL ownership, active lease exclusion, expiry recovery, stale-token fencing과 foreign/unowned isolation을 회귀 검증한다. `ledger_interval_sec`, retry `next_retry_at`, DB schema, ledger API와 dead-letter policy의 diff가 없는지 확인한다. +- 문서 조건: `docs/runtime-flows.md`와 `docs/interface-contract.md`가 empty claim INFO suppression, spy operator-only metrics, process-lifetime 및 restart-reset scope, fixed cadence 유지, sanitized writer exception과 loop survival을 명시하는지 직접 검사한다. +- 검사 방법: focused writer·messaging handler·daemon unit tests, InMemory parity tests, disposable PostgreSQL multi-instance tests, 전체 `pytest`, Ruff, `compileall`, `proofline validate`, `git diff --check`를 실행한다. PostgreSQL fixture가 없어서 skip되면 통과로 치환하지 않고 환경 미검증으로 보고한다. diff --git a/.proofline/identities.json b/.proofline/identities.json index 28811f2..9f8d54b 100644 --- a/.proofline/identities.json +++ b/.proofline/identities.json @@ -1,5 +1,5 @@ { "schema_version": 1, - "next_line_number": 12, - "next_ac_number": 40 + "next_line_number": 13, + "next_ac_number": 41 } diff --git a/.proofline/lines/line-0012/dcy-0012.md b/.proofline/lines/line-0012/dcy-0012.md new file mode 100644 index 0000000..846ab08 --- /dev/null +++ b/.proofline/lines/line-0012/dcy-0012.md @@ -0,0 +1,55 @@ +--- +id: "dcy-0012" +status: confirmed +--- + +# Ledger outbox idle poll 로그 억제와 관측 보존 + +## Problem + +`LedgerWriter.run_once()`는 처리할 `ledger_outbox` row가 없는 정상 idle poll에도 매번 `claim_result=empty` INFO 로그를 남긴다. daemon 기본 주기 1초와 결합되어 실제 원장 work item이 거의 없는 운영 환경에서도 대량의 정상 로그가 생성되고, claim 상실·retry·conflict·delivery failure 같은 조치 가능한 신호가 묻힌다. + +현재 `claim:empty` in-process counter는 증가하지만 writer 객체 밖의 운영 조회 표면에 연결되지 않는다. 따라서 empty INFO를 단순 삭제하면 로그 폭증은 해결되지만 empty poll 총량을 metric 또는 status에서 조회해야 한다는 Issue #97의 요구를 충족하지 못한다. 한편 polling cadence를 adaptive backoff나 event-driven wakeup으로 바꾸면 신규 item 처리 지연, process 간 알림, lease 만료·재claim에 대한 별도 계약이 필요하다. + +## Evidence + +- `E-001` GitHub Issue #97 (`https://github.com/fourblocks2020/trex-core/issues/97`)은 2026-09-16 00:00 KST부터 2026-09-18 21:31 KST까지 두 instance에서 `claim_result=empty` INFO 494,029줄이 발생했고 실제 신규 work item은 8건이었다고 기록한다. +- `E-002` 기준 revision `62e4f57c2b80ce28db87421c5e7b72a4316b8c5f`의 `trex_core/application/ledger_outbox.py`에서 `run_once()`는 `claim_pending()` 결과에 따라 `claim:empty` 또는 `claim:claimed` counter를 증가시키지만, empty/non-empty 구분 없이 INFO claim 로그를 남긴다. +- `E-003` 같은 파일의 `LedgerWriter.counters`는 process-local dict이며 이를 반환하는 status/RPC/exporter가 없다. 독립 read-only 조사도 현재 counter만으로는 운영 조회 가능성을 충족하지 못한다고 판정했다. +- `E-004` `trex_core/runtime/config.py`의 `ledger_interval_sec` 기본값은 1.0초이고 `trex_core/runtime/daemon.py::_ledger_loop()`는 매 interval마다 `run_once()`를 호출한다. writer 예외는 현재 로그 없이 삼켜진다. +- `E-005` `trex_core/application/ledger_outbox.py`의 batch claim INFO는 현재 `instance_key`, `claim_result`, `count`, `limit`만 남겨 correlation ID를 제공하지 않는다. targeted claim은 correlation ID를 남기며, claim lost 및 foreign/unowned item은 WARNING, retry/conflict/permanent/dead-letter outcome은 기존 counter와 로그 경로로 구분한다. +- `E-006` active `AC-0029`는 instance ownership, lease fencing, claim/outcome 관측과 secret 비노출을 소유하지만 idle poll 무로그와 운영 status 조회는 규정하지 않는다. 이번 capability는 `AC-0029`를 회귀 검증(`satisfy`)하면서 별도 delta AC를 생성(`create`)할 독립 운영 경계다. +- `E-007` focused baseline `uv run pytest -q tests/application/test_ledger_outbox.py tests/application/test_ledger_outbox_backoff.py tests/adapters/persistence/test_postgresql_ledger_outbox.py` 결과는 `20 passed, 16 skipped`였다. skip은 `TREX_POSTGRESQL_TEST_DSN`이 없는 PostgreSQL integration이다. +- `E-008` 저장소에는 범용 metrics exporter나 daemon status RPC가 없다. 기존 instance-scoped `trex.slots.req`는 `spy=true` 운영 조회와 `spy=false` 전략 조회를 구분하므로, operator 응답에만 additive process-lifetime ledger outbox metrics를 싣는 것이 새 transport/DB schema 없이 가능한 최소 surface다. + +## Scope + +- `S-001` `create`: 새 delta AC는 pending row가 없는 poll에서 `claim:empty`를 정확히 누적하되 기본 INFO 로그를 생성하지 않는 동작을 소유한다. 기본 `ledger_interval_sec=1.0`에서 24시간에 해당하는 `N=86,400`회의 `run_once()`를 실제 sleep 없이 두 writer(`instance-a`, `instance-b`)에 각각 호출한다. 각 writer의 `empty_polls_total`은 정확히 86,400이고 rendered INFO의 `claim_result=empty`는 0건이어야 한다. +- `S-002` `create`: instance-scoped `trex.slots.req`의 `spy=true` operator 응답에 top-level `ledger_outbox_metrics` snapshot을 additive로 노출한다. 정상 snapshot은 `available=true`, `scope=process_lifetime`, configured `instance_key`를 포함한다. `claims` object는 `empty_polls_total`, `claimed_batches_total`, `unavailable_total`, `lost_total`, `foreign_instance_total`을, `outcomes` object는 `completed_total`, `duplicate_completed_total`, `retryable_failure_total`, `conflict_blocked_total`, `permanent_failure_total`, `dead_lettered_total`을 누락 없이 0 이상의 정수로 반환한다. 값은 해당 daemon process의 configured instance writer만 집계하고 새 dict snapshot으로 반환한다. reader가 없거나 조회가 실패하면 `available=false`, `scope=process_lifetime`, configured `instance_key`, `reason=ledger_outbox_metrics_unavailable`을 반환하며 slots 조회 자체는 실패시키지 않는다. `spy=false` 전략 응답에는 `ledger_outbox_metrics` key 자체를 포함하지 않는다. +- `S-003` `create`: non-empty batch claim에서만 INFO를 남기고 non-zero `count`, `limit`, `instance_key`와 해당 batch의 non-empty correlation ID 목록을 보존한다. targeted claim은 기존 단일 correlation ID를 유지한다. empty poll은 correlation ID를 생성하거나 로그하지 않는다. +- `S-004` `satisfy AC-0029`: claim lost, foreign/unowned rejection, retryable, conflict, permanent failure, dead-letter와 성공/duplicate outcome의 severity, counter, instance ownership 및 lease fencing을 약화하지 않는다. 로그·metric·status에는 credential, account identifier, token, password, DSN, provider payload 또는 raw exception text를 포함하지 않는다. +- `S-005` `TrexDaemon._ledger_loop()`의 `ledger_writer.run_once()`가 예상 밖 예외를 던지면 ERROR 한 건에 `reason=ledger_writer_run_once_failed`, configured `instance_key`, exception class name인 `exc_type`만 남긴 뒤 다음 configured poll을 계속한다. 이 정책은 `_ledger_loop`에만 적용하며 raw exception message와 traceback/`exc_info`는 기록하지 않는다. sentinel 예외로 reason·type·loop survival과 secret 비노출을 함께 검증한다. +- `S-006` fixed `ledger_interval_sec` polling과 기존 retry `next_retry_at`, claim lease, ownership/filtering semantics는 변경하지 않는다. 신규 item은 daemon이 살아 있는 동안 enqueue 시점부터 다음 configured poll 기회에 처리 대상이 되어야 한다. +- `S-007` `docs/runtime-flows.md`와 관련 interface 문서를 idle logging, operator metrics, fixed cadence 결정에 맞게 동기화한다. + +## Out of Scope + +- `OS-001` adaptive idle backoff, PostgreSQL LISTEN/NOTIFY, event-driven wakeup, 별도 metrics exporter 및 durable metric persistence는 이번 Line에서 구현하지 않는다. 이들은 cross-process signaling과 별도 latency/SLO 설계가 필요한 후속 capability다. +- `OS-002` `ledger_outbox` schema, correlation ID 형식, lease TTL, retry/backoff 공식, dead-letter/conflict 정책과 portfolio-hub API 계약은 변경하지 않는다. +- `OS-003` AC-0029 원문, 승인된 기존 REQ/Discovery bytes와 기존 lifecycle status는 변경하지 않는다. +- `OS-004` production daemon 재기동, production PostgreSQL row 조작, 실제 원장 delivery와 운영 로그 삭제는 수행하지 않는다. + +## Risks and Unknowns + +### Risks + +- `R-001` empty와 non-empty 분기를 잘못 배치하면 실제 claim INFO까지 사라질 수 있다. empty/no-log와 non-empty/identity-log를 별도 회귀로 검증한다. +- `R-002` process-lifetime metric은 daemon 재시작 시 0으로 초기화된다. response에 scope를 명시하고 durable total로 오인하지 않도록 문서화한다. +- `R-003` operator metric을 strategy 응답에도 싣거나 mutable counters dict를 그대로 반환하면 불필요한 내부 정보 노출 또는 동시 변경 위험이 생긴다. `spy=true`에서만 정수 snapshot을 새 dict로 반환한다. +- `R-004` daemon exception 로그에 raw exception text나 traceback을 넣으면 DSN·credential이 노출될 수 있다. sentinel 예외로 sanitized logging과 loop survival을 함께 검증한다. +- `R-005` PostgreSQL fixture가 없으면 real-backend multi-instance/lease 회귀는 skip된다. 해당 gap을 명시하고 기존 gated integration selector를 유지한다. + +### Assumptions + +- `A-001` Issue #97의 “metric 또는 status에서 조회”는 daemon process lifetime 동안 instance별 누적값을 operator의 `spy=true` slots RPC/CLI JSON 응답에서 조회하는 것으로 충족한다. durable historical aggregation은 요구하지 않는다. +- `A-002` 허용 가능한 신규 item 지연은 기존 configured `ledger_interval_sec`의 다음 poll 기회까지다. cadence를 유지하므로 이번 변경이 추가 지연을 만들지 않는다. diff --git a/.proofline/lines/line-0012/evidence/activity-log.md b/.proofline/lines/line-0012/evidence/activity-log.md new file mode 100644 index 0000000..7122a23 --- /dev/null +++ b/.proofline/lines/line-0012/evidence/activity-log.md @@ -0,0 +1,35 @@ +# Activity Log — line-0012 + +## Specification approval + +- Issue: `fourblocks2020/trex-core#97` +- Base revision: `62e4f57c2b80ce28db87421c5e7b72a4316b8c5f` +- Branch: `feat/issue-97-outbox-idle-observability` +- Discovery `dcy-0012` was confirmed by the user after independent review PASS. +- Confirmed Discovery digest after lifecycle transition: `8a18a9d1a22d92f13f59c2b6d0ce942a3dc24d0959cf8943d1ea843a8e7e536f`. +- Specification review PASS covered `REQ-0012`, create `AC-0040`, and unchanged satisfy target `AC-0029`. +- The user explicitly approved `REQ-0012` and `AC-0040/AC-0029` from approval packet SHA-256 `f1460df034b635174f8a64d59ceec1c9f5d2c3c4f818ab4b939550e455121666`. +- Lifecycle result: `REQ-0012=approved`, `AC-0040=active`; `AC-0029` remains active and unchanged. +- No Normative Design Documents are part of the approval baseline. + +## Implementation plan + +1. Suppress empty-claim INFO while retaining process-lifetime counters and exact non-empty batch correlation IDs. +2. Add immutable operator metrics snapshots to spy-only slots RPC responses. +3. Add sanitized `_ledger_loop` exception observability and prove loop survival. +4. Synchronize runtime/interface documentation and run focused, full, static, ProofLine, and independent verification. +5. Deliver a pull request without production deployment or live remediation. + +## Implementation and verification + +- idle empty claim INFO를 제거하고 instance-local process-lifetime metric snapshot을 추가했다. +- `trex.slots.req`의 `spy=true` 응답에 metrics를 연결하고 `spy=false` 응답은 기존 shape를 유지했다. +- `_ledger_loop` unexpected writer exception을 sanitized ERROR로 기록하고 loop를 유지하도록 했다. +- `docs/runtime-flows.md`, `docs/interface-contract.md`를 구현 계약과 동기화했다. +- Focused tests: `42 passed, 16 skipped`; PostgreSQL fixture 부재로 16건은 환경 미검증이다. +- Full tests: `1267 passed, 98 skipped`; skip에는 PostgreSQL fixture 의존 테스트가 포함된다. +- Ruff, `compileall`, `proofline validate`, `git diff --check`: 통과. +- Base revision sabotage probes: old idle logging, missing snapshot, missing RPC wiring, silent daemon exception에 대해 `4 failed`로 변경 검출을 확인했다. +- 독립 구현 검토: PASS; production composition wiring을 포함해 material blocker 없음. +- 미실행: production daemon lifecycle, broker 운영, production PostgreSQL 변경 또는 재기동. +- 다음 행동: commit, remote branch push, PR 생성 및 exact head CI 확인. diff --git a/.proofline/lines/line-0012/line-0012.md b/.proofline/lines/line-0012/line-0012.md new file mode 100644 index 0000000..5d5968a --- /dev/null +++ b/.proofline/lines/line-0012/line-0012.md @@ -0,0 +1,4 @@ +--- +id: "line-0012" +status: discovery +--- diff --git a/.proofline/lines/line-0012/req-0012.md b/.proofline/lines/line-0012/req-0012.md new file mode 100644 index 0000000..7fd9cf9 --- /dev/null +++ b/.proofline/lines/line-0012/req-0012.md @@ -0,0 +1,43 @@ +--- +id: "req-0012" +status: approved +discovery: "dcy-0012" +criteria: + create: + - "ac-0040" + update: [] + retire: [] + satisfy: + - "ac-0029" +--- + +# Ledger outbox idle poll 로그 억제와 관측 보존 + +## Objective + +정상 idle 상태의 ledger outbox poll이 초 단위 INFO 로그를 만들지 않게 하면서, process-lifetime claim·outcome 통계를 operator가 instance별로 조회할 수 있게 한다. 실제 work와 failure는 correlation 가능한 sanitized observability를 유지하고 기존 lease·ownership·retry 동작은 변경하지 않는다. + +## Scope + +- pending row가 없는 `LedgerWriter.run_once()`는 `claim:empty`를 누적하지만 INFO claim log를 남기지 않는다. +- non-empty batch claim INFO는 configured `instance_key`, non-zero `count`, `limit`, 해당 batch의 non-empty correlation ID를 repository claim 순서와 같은 `correlation_ids` JSON string array로 제공한다. +- `trex.slots.req`의 `spy=true` 응답에 top-level `ledger_outbox_metrics` process-lifetime snapshot을 additive로 제공한다. `spy=false` 응답에는 해당 key를 포함하지 않는다. +- metrics reader가 없거나 조회에 실패해도 slots 조회는 성공하며 stable unavailable snapshot을 반환한다. +- `_ledger_loop()`의 예상 밖 writer exception은 stable reason, configured `instance_key`, exception class name만 ERROR로 남기고 다음 poll을 계속한다. +- `AC-0029`의 ownership, active lease exclusion, stale outcome fencing, foreign/unowned isolation, retry 및 terminal outcome 계약을 회귀 검증한다. +- `docs/runtime-flows.md`와 `docs/interface-contract.md`를 idle suppression, operator-only metrics, process-lifetime scope, fixed cadence와 sanitized writer exception contract에 맞게 동기화한다. + +## Constraints + +- metrics snapshot은 해당 daemon process의 configured instance writer만 집계하는 process-lifetime 값이며 process restart 시 0부터 시작한다. +- counter는 누락 없는 0 이상의 정수이고 호출자가 내부 mutable state를 변경할 수 없는 새 snapshot이어야 한다. +- log, metric, status에는 credential, account identifier, token, password, DSN, provider payload, raw exception message 또는 traceback을 포함하지 않는다. +- fixed `ledger_interval_sec`, retry `next_retry_at`, claim lease, ownership/filtering, persistence schema와 ledger API contract는 변경하지 않는다. +- PostgreSQL multi-instance verification은 disposable fixture가 있을 때 실행하며 fixture 부재로 skip되면 환경 미검증으로 명시한다. + +## Non-Goals + +- adaptive idle backoff 또는 event-driven wakeup을 추가하지 않는다. +- 범용 metrics exporter나 신규 transport/status RPC를 만들지 않는다. +- portfolio/business-state, broker order/fill, correlation-ID format, dead-letter policy를 변경하지 않는다. +- production deployment, live database migration, backfill 또는 remediation을 수행하지 않는다. diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 1749117..2bde924 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -4,7 +4,7 @@ - Baseline Seed: .stage-pilot/bootstrap/baseline.yaml - Status: draft - Owner: project maintainers -- Last Updated (KST): 2026-09-19 (portfolio-hub Issue #32 venue provenance transport mapping) +- Last Updated (KST): 2026-09-20 (trex-core Issue #97 ledger outbox idle observability) - Source Discovery / Batch: bootstrap-baseline (pre-Discovery); HTTP API = dcy-002 / BAT-002; operator broker truth = dcy-037 / BAT-057; ledger venue transport = portfolio-hub ProofLine req-0009 - Primary Runtime: daemon (HTTP API) + library (slot 도메인) @@ -63,7 +63,7 @@ | 패턴 | 표면 | 요청 속성 | 응답 | |---|---|---|---| -| 슬롯 RPC | `trex.slots.req` | `correlation_id`, `reply_to=amq.rabbitmq.reply-to` | reply-to 큐로 `{slots[], degraded, reason, message}` (REQ-073/074, BAT-035) | +| 슬롯 RPC | `trex.slots.req` | `correlation_id`, `reply_to=amq.rabbitmq.reply-to` | reply-to 큐로 `{slots[], degraded, reason, message, ledger_outbox_metrics?}` (`ledger_outbox_metrics`는 `spy=true` 전용, REQ-073/074, AC-0040) | | intent RPC | `trex.intent.req` | `correlation_id` = intent 멱등키, `reply_to=amq.rabbitmq.reply-to`, `delivery_mode=2` | reply-to 큐로 `{intent_id, status:admitted|rejected, slot_id?, reason?}`. retryability·broker attempt·KIS failure 정보는 strategy 응답에 포함하지 않는다. 체결/종료는 응답 아닌 이벤트로 | | update_thresholds | `trex.intent.req` | `action="update_thresholds"`, `reply_to` | reply-to 큐로 `{status:updated|ignored, slot_id, reason?}` (REQ-040) | | 이벤트 pub/sub | `trex.events` (topic) | routing key `{event_type}.{strategy_id}.{symbol}` | 전략 큐가 자기 `strategy_id` 패턴으로 바인딩 | @@ -162,6 +162,17 @@ - `spy=false`(기본): holding 슬롯 응답 후 `WatchdogStore.update(slot_id)` 호출 — 전략 폴링 = 생존 신호. - `spy=true`: watchdog 갱신 건너뜀 — 운영자 CLI(`universe`/`slots`) 가 항상 사용. +#### `spy=true` ledger outbox metrics (AC-0040) + +- `ledger_outbox_metrics`는 `spy=true` 운영자 응답의 top-level additive field다. `spy=false` 전략 응답에는 key 자체를 포함하지 않는다. +- 정상 snapshot은 `available=true`, `scope="process_lifetime"`, configured `instance_key`와 다음 누적 정수를 제공한다. + - `claims`: `empty_polls_total`, `claimed_batches_total`, `unavailable_total`, `lost_total`, `foreign_instance_total` + - `outcomes`: `completed_total`, `duplicate_completed_total`, `retryable_failure_total`, `conflict_blocked_total`, `permanent_failure_total`, `dead_lettered_total` +- 각 값은 해당 daemon process의 configured instance writer에서만 누적하며 0 이상의 정수다. snapshot은 내부 mutable counter와 분리된 새 값이고 daemon restart 때 0으로 초기화된다. +- metrics reader가 없거나 조회에 실패하면 slots 조회는 계속 성공하고 `available=false`, `scope="process_lifetime"`, configured `instance_key`, `reason="ledger_outbox_metrics_unavailable"`만 반환한다. raw exception text는 응답에 포함하지 않는다. +- ledger writer는 fixed `ledger_interval_sec` cadence를 유지한다. empty claim은 `empty_polls_total`만 증가시키고 INFO를 남기지 않는다. non-empty batch INFO만 `instance_key`, non-zero `count`, `limit`, repository claim 순서의 `correlation_ids` JSON string array를 포함한다. +- `_ledger_loop`의 예상 밖 writer exception은 `reason=ledger_writer_run_once_failed`, configured instance, exception class name만 ERROR로 남긴다. raw exception message, traceback과 `exc_info`는 금지하며 loop는 다음 configured poll을 계속한다. + ### 운영자 CLI ↔ RPC 매핑 (dcy-016 / REQ-032) 운영자 CLI 명령은 데몬 RabbitMQ RPC 로 통일된다(단일 writer admission 보존, DB 직접 쓰기 없음). 인스턴스 라우팅: `--config `(canonical, REQ-048) 또는 `--instance `(단축·override), 미지정 시 `default`. diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index 7e7032b..6712590 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -50,7 +50,7 @@ 1. 전략 모듈 또는 운영자가 RPC 요청을 `trex.slots.req` 큐로 publish — payload 에 `filter`(status 필터) + `spy`(워치독 억제 여부), properties 에 `correlation_id` + `reply_to`. 2. `trex-slots-rpc` 스레드가 메시지 수신 → `make_slots_rpc_handler(slot_repo, ...)` 로 슬롯 목록 산출. 3. `spy=false`(기본) 면 holding 슬롯 `WatchdogStore.update(slot_id)` 호출(전략 폴링 = 생존 신호, REQ-034). -4. 응답 `{slots[], degraded, reason, message}` 를 `reply_to` 큐로 publish → manual ack. +4. 응답 `{slots[], degraded, reason, message}` 를 `reply_to` 큐로 publish → manual ack. `spy=true` 운영자 응답에는 process-lifetime `ledger_outbox_metrics` snapshot을 top-level에 추가하고, `spy=false` 전략 응답에는 해당 key를 노출하지 않는다. 5. read-only 이므로 dedup 불요; 재요청 시 동일 응답을 다시 생성한다. ## Flow: `매수/매도 intent admission RPC` (RabbitMQ `trex.intent.req`, BAT-007) @@ -314,6 +314,8 @@ timeout 미지정·취소 실패로 잔여 고착된 pending 슬롯을 처분: 4. 터미널 실패 가시화(dcy-015/REQ-018): `delivery_failed`(permanent/dead-letter)는 `WriterStats.delivery_failed` 카운터 + WARN 로그 + 구분 status 로 노출되어 정상 `pending` 과 구별된다(무성 영속 실패 금지). dead-letter 는 폐기가 아니라 운영자 재처리 가능 상태이며 `find_ledger_gap_slots`(REQ-055)·skip 가시화와 연계해 누락을 추적한다. 5. 백오프 파라미터는 env 설정: `TREX_LEDGER_RETRY_BASE_SEC`(기본 1.0)/`TREX_LEDGER_RETRY_CAP_SEC`(기본 300)/`TREX_LEDGER_MAX_ATTEMPTS`(기본 8). 6. portfolio-hub 장기 장애에도 outbox 가 적재되어 무유실. 재시작 시 PostgreSQL `ledger_outbox.status='pending'` 행을 그대로 이어 처리한다(터미널 `delivery_failed`/`conflict_blocked` 는 제외). +7. 데몬은 configured `ledger_interval_sec`(기본 1초)의 fixed cadence를 유지한다. pending row가 없는 정상 idle poll은 process-local `claim:empty` counter만 증가시키고 INFO 로그를 남기지 않는다. non-empty batch만 INFO를 남기며 configured `instance_key`, non-zero `count`, `limit`, repository claim 순서의 `correlation_ids` JSON string array를 포함한다. +8. process-lifetime counter는 daemon restart 때 0으로 초기화되며 DB나 별도 exporter에 영속화하지 않는다. `trex.slots.req`의 `spy=true` 응답에서 claim/outcome totals를 조회할 수 있다. writer 호출이 예상 밖 예외를 던지면 `_ledger_loop`는 `reason=ledger_writer_run_once_failed`, configured instance와 exception class name만 ERROR로 남기고 다음 poll을 계속한다. raw exception message, traceback과 `exc_info`는 기록하지 않는다. ## Flow: `구조화 로깅 / 관측성` (BAT-045·046 / dcy-029, REQ-084~088) diff --git a/tests/application/test_ledger_outbox.py b/tests/application/test_ledger_outbox.py index 64fc05c..1f3353c 100644 --- a/tests/application/test_ledger_outbox.py +++ b/tests/application/test_ledger_outbox.py @@ -1,6 +1,7 @@ """LedgerOutbox + LedgerWriter direct unit tests (REQ-019).""" from __future__ import annotations +import logging from dataclasses import replace from datetime import UTC, datetime, timedelta @@ -132,6 +133,80 @@ def test_writer_exposes_claim_and_outcome_counters(): assert writer.counters["outcome:completed"] == 1 +def test_idle_day_is_silent_and_counted_per_instance(caplog): + writers = [ + LedgerWriter( + InMemoryLedgerOutbox(instance_key=instance_key), + InMemoryLedger(), + instance_key=instance_key, + ) + for instance_key in ("instance-a", "instance-b") + ] + + with caplog.at_level(logging.INFO, logger="trex_core.application.ledger_outbox"): + for writer in writers: + for _ in range(86_400): + writer.run_once() + + assert [ + writer.metrics_snapshot()["claims"]["empty_polls_total"] + for writer in writers + ] == [86_400, 86_400] + assert "claim_result=empty" not in caplog.text + assert "correlation_ids" not in caplog.text + + +def test_nonempty_claim_log_has_ordered_json_correlation_ids(caplog): + outbox = InMemoryLedgerOutbox(instance_key="inst-a") + writer = LedgerWriter(outbox, InMemoryLedger(), instance_key="inst-a") + first = outbox.enqueue(_trade(execution_id="e-1")) + second = outbox.enqueue(_trade(execution_id="e-2")) + + with caplog.at_level(logging.INFO, logger="trex_core.application.ledger_outbox"): + writer.run_once(limit=2) + + claim_logs = [ + record.getMessage() + for record in caplog.records + if "ledger outbox claim:" in record.getMessage() + ] + assert claim_logs == [ + "ledger outbox claim: instance_key=inst-a claim_result=claimed " + f"count=2 limit=2 correlation_ids=[\"{first.correlation_id}\"," + f"\"{second.correlation_id}\"]" + ] + + +def test_metrics_snapshot_is_complete_and_detached(): + outbox = InMemoryLedgerOutbox(instance_key="inst-a") + writer = LedgerWriter(outbox, InMemoryLedger(), instance_key="inst-a") + + snapshot = writer.metrics_snapshot() + + assert snapshot == { + "available": True, + "scope": "process_lifetime", + "instance_key": "inst-a", + "claims": { + "empty_polls_total": 0, + "claimed_batches_total": 0, + "unavailable_total": 0, + "lost_total": 0, + "foreign_instance_total": 0, + }, + "outcomes": { + "completed_total": 0, + "duplicate_completed_total": 0, + "retryable_failure_total": 0, + "conflict_blocked_total": 0, + "permanent_failure_total": 0, + "dead_lettered_total": 0, + }, + } + snapshot["claims"]["empty_polls_total"] = 999 + assert writer.metrics_snapshot()["claims"]["empty_polls_total"] == 0 + + @pytest.mark.parametrize("instance_key", ["", " "]) def test_writer_rejects_explicit_empty_instance_key(instance_key): outbox = InMemoryLedgerOutbox(instance_key="inst-a") diff --git a/tests/application/test_slots_handler.py b/tests/application/test_slots_handler.py index bea909f..083654a 100644 --- a/tests/application/test_slots_handler.py +++ b/tests/application/test_slots_handler.py @@ -207,6 +207,66 @@ def test_spy_true_no_watchdog_update(): assert watchdog.last_seen(sid) is None # still not updated (spy=True) +def test_spy_response_exposes_ledger_outbox_metrics(): + class _MetricsReader: + def metrics_snapshot(self): + return { + "available": True, + "scope": "process_lifetime", + "instance_key": "default", + "claims": { + "empty_polls_total": 7, + "claimed_batches_total": 1, + "unavailable_total": 0, + "lost_total": 0, + "foreign_instance_total": 0, + }, + "outcomes": { + "completed_total": 1, + "duplicate_completed_total": 0, + "retryable_failure_total": 0, + "conflict_blocked_total": 0, + "permanent_failure_total": 0, + "dead_lettered_total": 0, + }, + } + + repo, *_ = _wire() + handler = make_slots_rpc_handler( + repo, + ledger_outbox_metrics_reader=_MetricsReader(), + ) + + spy_payload = handler(_req(spy=True)).payload + strategy_payload = handler(_req(spy=False)).payload + + assert spy_payload["ledger_outbox_metrics"]["claims"]["empty_polls_total"] == 7 + assert "ledger_outbox_metrics" not in strategy_payload + + +def test_spy_response_reports_metrics_unavailable_without_failing_slots(): + class _FailingReader: + def metrics_snapshot(self): + raise RuntimeError("SECRET_METRICS_READER_FAILURE") + + repo, *_ = _wire() + for reader in (None, _FailingReader()): + handler = make_slots_rpc_handler( + repo, + instance_key="default", + ledger_outbox_metrics_reader=reader, + ) + payload = handler(_req(spy=True)).payload + + assert len(payload["slots"]) == 2 + assert payload["ledger_outbox_metrics"] == { + "available": False, + "scope": "process_lifetime", + "instance_key": "default", + "reason": "ledger_outbox_metrics_unavailable", + } + + # REQ-073 AC-4: spy=False (default) → holding watchdog updated def test_spy_false_updates_watchdog(): repo, svc, _, _, watchdog, handler = _wire() diff --git a/tests/runtime/test_daemon_ledger_loop_exception_logging.py b/tests/runtime/test_daemon_ledger_loop_exception_logging.py new file mode 100644 index 0000000..656c82c --- /dev/null +++ b/tests/runtime/test_daemon_ledger_loop_exception_logging.py @@ -0,0 +1,47 @@ +"""Ledger writer loop failures remain visible without leaking exception text.""" +from __future__ import annotations + +import logging +import threading +from typing import Any + +from trex_core.runtime.daemon import Daemon + +_SENTINEL = "SECRET-LEDGER-DSN-PAYLOAD-DO-NOT-LOG" + + +def test_ledger_loop_logs_sanitized_error_and_continues(caplog): + call_count = 0 + stop_event = threading.Event() + + class _Writer: + def run_once(self): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise RuntimeError(f"dsn=postgresql://user:{_SENTINEL}@host/db") + stop_event.set() + + class _Config: + ledger_interval_sec = 0.001 + instance_key = "instance-a" + + daemon: Any = object.__new__(Daemon) + daemon.stop = stop_event + daemon.config = _Config() + daemon.ledger_writer = _Writer() + + with caplog.at_level(logging.ERROR, logger="trex_core.runtime.daemon"): + thread = threading.Thread(target=daemon._ledger_loop, daemon=True) + thread.start() + thread.join(timeout=2.0) + + assert not thread.is_alive() + assert call_count >= 2 + assert len(caplog.records) == 1 + record = caplog.records[0] + assert "reason=ledger_writer_run_once_failed" in record.getMessage() + assert "instance_key=instance-a" in record.getMessage() + assert "exc_type=RuntimeError" in record.getMessage() + assert _SENTINEL not in caplog.text + assert record.exc_info is None diff --git a/trex_core/application/ledger_outbox.py b/trex_core/application/ledger_outbox.py index cf043ed..30cdc06 100644 --- a/trex_core/application/ledger_outbox.py +++ b/trex_core/application/ledger_outbox.py @@ -23,13 +23,14 @@ critical path while making ledger writes eventually-consistent and idempotent. """ from __future__ import annotations +import json import logging import random from collections.abc import Callable from dataclasses import dataclass, replace from datetime import datetime, timedelta, timezone from enum import Enum -from typing import Optional, Protocol +from typing import Any, Optional, Protocol from trex_core.application.alarm_emitter import AlarmEmitter from trex_core.application.ledger_port import ( @@ -223,6 +224,41 @@ class LedgerWriter: def _count(self, key: str) -> None: self.counters[key] = self.counters.get(key, 0) + 1 + def metrics_snapshot(self) -> dict[str, Any]: + """Return a detached process-lifetime observability snapshot.""" + return { + "available": True, + "scope": "process_lifetime", + "instance_key": self._instance_key, + "claims": { + "empty_polls_total": self.counters.get("claim:empty", 0), + "claimed_batches_total": self.counters.get("claim:claimed", 0), + "unavailable_total": self.counters.get("claim:unavailable", 0), + "lost_total": self.counters.get("claim:lost", 0), + "foreign_instance_total": self.counters.get( + "claim:foreign_instance", 0 + ), + }, + "outcomes": { + "completed_total": self.counters.get("outcome:completed", 0), + "duplicate_completed_total": self.counters.get( + "outcome:duplicate_completed", 0 + ), + "retryable_failure_total": self.counters.get( + "outcome:retryable_failure", 0 + ), + "conflict_blocked_total": self.counters.get( + "outcome:conflict_blocked", 0 + ), + "permanent_failure_total": self.counters.get( + "outcome:permanent_failure", 0 + ), + "dead_lettered_total": self.counters.get( + "outcome:dead_lettered", 0 + ), + }, + } + def enqueue_fill(self, trade: LedgerTrade) -> LedgerWorkItem: return self._outbox.enqueue(trade) @@ -231,13 +267,20 @@ class LedgerWriter: items = self._outbox.claim_pending(limit) claim_result = "claimed" if items else "empty" self._count(f"claim:{claim_result}") - _log.info( - "ledger outbox claim: instance_key=%s claim_result=%s count=%d limit=%d", - self._instance_key, - claim_result, - len(items), - limit, - ) + if items: + correlation_ids = json.dumps( + [item.correlation_id for item in items], + ensure_ascii=True, + separators=(",", ":"), + ) + _log.info( + "ledger outbox claim: instance_key=%s claim_result=claimed " + "count=%d limit=%d correlation_ids=%s", + self._instance_key, + len(items), + limit, + correlation_ids, + ) for item in items: if not self._validate_instance(item): continue diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index dc8f62f..f313434 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -259,6 +259,7 @@ def make_slots_rpc_handler( watchdog_store: Optional[WatchdogStore] = None, watchdog_timeout_sec: int = 120, ledger_outbox=None, + ledger_outbox_metrics_reader=None, max_active_slots: Optional[int] = None, strategy_state_store=None, broker_quantity_reader: Optional[SlotsBrokerQuantityReader] = None, @@ -512,6 +513,23 @@ def make_slots_rpc_handler( ) if health_payload: base_payload["broker_balance_health"] = health_payload + if spy: + unavailable_metrics = { + "available": False, + "scope": "process_lifetime", + "instance_key": instance_key, + "reason": "ledger_outbox_metrics_unavailable", + } + if ledger_outbox_metrics_reader is None: + base_payload["ledger_outbox_metrics"] = unavailable_metrics + else: + try: + metrics = ledger_outbox_metrics_reader.metrics_snapshot() + if not isinstance(metrics, dict): + raise TypeError("metrics snapshot must be a dict") + base_payload["ledger_outbox_metrics"] = metrics + except Exception: # noqa: BLE001 - status response must stay available + base_payload["ledger_outbox_metrics"] = unavailable_metrics if warn_store is not None: base_payload["excluded_slots"] = { "reconcile_quantity_mismatch": excluded_count, diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index 46605db..e6d83b9 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -452,8 +452,13 @@ class Daemon: while not self.stop.wait(self.config.ledger_interval_sec): try: self.ledger_writer.run_once() - except Exception: # noqa: BLE001,S112 - continue + except Exception as exc: # noqa: BLE001 - loop must survive transient errors + _log.error( + "ledger writer loop error: reason=ledger_writer_run_once_failed " + "instance_key=%s exc_type=%s", + self.config.instance_key, + type(exc).__name__, + ) def _intent_rpc_loop(self) -> None: transport = _build_transport(self.config) @@ -486,6 +491,7 @@ class Daemon: watchdog_store=self.watchdog, watchdog_timeout_sec=self.config.safety_sell.watchdog_timeout_sec, ledger_outbox=self.outbox, + ledger_outbox_metrics_reader=self.ledger_writer, max_active_slots=self.config.max_active_slots, strategy_state_store=self.strategy_state_store, # dcy-028 C-1 broker_quantity_reader=self.account_snapshot_service,