diff --git a/docs/interface-contract.md b/docs/interface-contract.md index 414cfcf..05b00df 100644 --- a/docs/interface-contract.md +++ b/docs/interface-contract.md @@ -74,11 +74,14 @@ ``` { "filter": "all" | "inactive" | "pending" | "holding" | "error", # default "all" "spy": true | false, # default false; true = watchdog 갱신 안 함(운영자 조회) - "strategy": str # optional (REQ-116 BAT-055) — exact-match on strategy_name + "strategy": str, # optional (REQ-116 BAT-055) — exact-match on strategy_name + "strategy_instance_id": str # required non-empty string when spy=false; ignored for spy=true filtering } ``` - `filter`: 서버 측 슬롯 status 필터. `"inactive"` 는 buyable 슬롯(universe에 바인딩된 매수 가능 슬롯). 미인식 값은 `"all"` 로 처리. - `spy=true`: 응답에 포함된 holding 슬롯의 워치독 `last_seen_at` 을 갱신하지 않는다(운영자 CLI 조회가 전략 생존 신호로 오인되지 않도록 — REQ-034 CHG-01). +- `spy=false`: `strategy_instance_id`는 required non-empty string이다. 누락·비문자열·빈 값은 `{slots:[], available_slots:null, degraded:false, reason:"bad_request:strategy_instance_id", message:"strategy_instance_id must be a non-empty string"}`를 반환하며 watchdog을 갱신하지 않는다. +- strategy-state visibility: `spy=false`는 exact requested entry 하나를 `strategy_state: {version,data,updated_at}`로만 반환하고 missing entry는 생략한다. `spy=true`는 supplied ID와 무관하게 complete `{strategy_instance_id: {version,data,updated_at}}` map을 반환한다. - `strategy` (REQ-116 additive): 지정 시 `strategy_name == strategy` 인 슬롯 행만 반환. 대소문자 exact-match. 미지정 시 기존과 완전 동일. `filter` 와 AND 조합. 매칭 없으면 `slots=[]`(에러 없음). `available_slots` 는 strategy 필터와 무관하게 인스턴스 전체 기준으로 계산(REQ-073 AC-6 불변). 코어가 reply-to 큐로 보내는 응답 payload: @@ -524,21 +527,21 @@ get_current_price(symbol: str, market: str = "J") -> Optional[Decimal] - pika BlockingConnection 은 thread-safe 가 아니므로 consumer/publisher 마다 자기 connection/channel 을 가져야 한다. - 토폴로지 선언은 멱등하지만 attribute(durable, args) 가 기존 큐와 다르면 RabbitMQ 가 PRECONDITION_FAILED 로 거부 → batch 마이그레이션 시 큐 재선언 절차 필요. -## Interface: `trex.slots.strategy_state.set` RPC (BAT-042 / dcy-028, REQ-080) +## Interface: `trex.slots.strategy_state.set` RPC (REQ-0009) - Type: external-service (strategy daemon ↔ core write RPC) - Consumers / Producers: 전략 데몬(produce) ↔ core 핸들러(consume). per-instance 큐(`{base}.{instance_key}`), direct reply-to, JSON. -- Purpose: 슬롯의 opaque strategy-state 를 occupancy_seq+version CAS 로 write. read 는 `trex.slots.req` 에 piggyback(별도 RPC 아님). +- Purpose: 슬롯별 opaque strategy-instance state entry를 version CAS로 write. read는 `trex.slots.req`에 piggyback(별도 RPC 아님). ### Inputs -- `{slot_id, strategy_id, expected_version:int, expected_occupancy_seq:int, data:str}`. `correlation_id = "{slot_id}:{strategy_id}:{expected_version}"`(멱등), delivery persistent. `data` 는 opaque UTF-8(core 무해석), byte size cap(기본 4KB ceiling, 운영 권장 1~2KB). +- `{slot_id, strategy_instance_id, expected_version:int, data:str}`. `strategy_instance_id`는 required non-empty string이며 `correlation_id`는 저장 멱등성 key가 아니다. `data`는 opaque UTF-8(core 무해석), byte size cap(기본 4KB ceiling, 운영 권장 1~2KB). ### Outputs / Outcomes -- 성공: `{ok:true, slot_id, strategy_id, version:}`. -- 실패 reason 전체 집합: `occupancy_changed`(+`occupancy_seq`) / `version_conflict`(+`version`) / `slot_inactive` / `state_too_large` / `bad_request:`. 소비자는 미인식 reason 에 fail-closed. +- 성공: `{ok:true, slot_id, strategy_instance_id, version:}`. +- 실패 reason 전체 집합: `version_conflict`(+`version`) / `slot_inactive` / `state_too_large` / `bad_request:`. 소비자는 미인식 reason에 fail-closed. ### Compatibility Rules -- read 표면(`trex.slots.req`)의 각 슬롯 행에 additive 로 `occupancy_seq`(int)와, 보유 시 `strategy_state` 맵(`{:{version,data,updated_at}}`)이 실린다(REQ-073/074 CHG). 기존 소비자는 unknown 필드 무시로 하위호환. `strategy_state` 는 strategy_* reasoning 필드(strategy_id 등, REQ-074 AC-5 제외 대상)와 구분되는 opaque 상태 맵이다. +- slot strategy-state는 occupancy와 독립된 slot-scoped `data[strategy_instance_id] = {version,data,updated_at}` map이다. close/reopen은 entry를 삭제하거나 version을 초기화하지 않는다. Entry에는 `strategy_id`나 `occupancy_seq` metadata가 없으며, Slot 자체의 `occupancy_seq` semantics는 별도로 유지된다. ## Shared Constraints diff --git a/docs/runtime-flows.md b/docs/runtime-flows.md index fe4a41c..68b97cd 100644 --- a/docs/runtime-flows.md +++ b/docs/runtime-flows.md @@ -34,11 +34,12 @@ 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개 백그라운드 스레드 동시 기동 +3. start: 8개 백그라운드 스레드 동시 기동 - `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) + - `trex-slots-rpc` :: `trex.slots.req` 소비자 (spy=false 시 holding 슬롯 워치독 `last_seen_at` 갱신 REQ-034; `safety_sell_enabled=false` 시 `trex-safety-sell` 미기동 BAT-036) + - `trex-strategy-state-set` :: `trex.slots.strategy_state.set` 소비자 (slot-scoped strategy-instance state version CAS) - `trex-control-rpc` :: `trex.control.req` 소비자 — pause/resume/list + kill_switch/resume_kill_switch (REQ-032/043) - `trex-snapshot-scheduler` :: 일일 portfolio-hub 스냅샷 refresh + **universe 슬롯 reconcile**(REQ-058) + holding 불일치 reconcile(REQ-062) - `trex-safety-sell` :: 자율 안전 매도 트리거 루프 (forced_close / stop-loss·take-profit / 장 마감, REQ-037/038/039) @@ -138,7 +139,7 @@ `trex.slots.req` 메시지 처리 (`make_slots_rpc_handler`). -1. 요청 payload 디코드: `{filter ∈ {all, inactive, pending, holding, error}, spy: bool}`. +1. 요청 payload 디코드: `{filter ∈ {all, inactive, pending, holding, error}, spy: bool, strategy_instance_id?: str}`. `spy=false` 전략 요청은 non-empty string `strategy_instance_id`가 필수이며, malformed 값은 slots/state 노출과 watchdog 갱신 없이 `bad_request:strategy_instance_id` 기존 envelope으로 종료한다. `spy=true` 운영자 요청은 ID를 요구하거나 supplied ID로 결과를 필터하지 않는다. 2. `slot_repo.list_slots(instance_key)` → status 기준 서버 필터 적용(`inactive`/`pending`/`holding`/`error`; 미인식 = `all`). 3. `spy=false` 면 `slot_repo.list_holding_slots()` 의 모든 holding 슬롯에 `WatchdogStore.update(slot_id)`. 4. 각 슬롯에 `slot_to_dict(s, snapshot_store=..., watchdog_store=..., pause_store=..., watchdog_timeout_sec=...)` 적용 후 broker holdings snapshot 오버레이를 적용해 병합 row 산출: @@ -149,7 +150,7 @@ - 추정 현금(REQ-029 OQ-4): `cash_estimated=True` + `cash_balance += pending ledger delta`(best-effort) - 워치독: `watchdog: {expired, last_seen}` inline (WatchdogStore) - 오버레이: `paused` (PauseStore, REQ-032) - - strategy 필드 없음 (REQ-074 AC-5) + - strategy-state: `spy=false`면 요청 `strategy_instance_id`와 정확히 일치하는 `{version,data,updated_at}` entry 하나만(없으면 생략), `spy=true`면 `{strategy_instance_id: {version,data,updated_at}}` 전체 map 5. degraded 상태 동봉: `SnapshotStore.degraded_state()` → `{degraded, reason, message}`. 6. 응답: `{slots[], degraded, reason, message}` — `slots` 는 항상 배열, null 금지. @@ -346,17 +347,15 @@ timeout 미지정·취소 실패로 잔여 고착된 pending 슬롯을 처분: 6. slot_transition_log 에 전이 기록 1행을 append 하고 갱신된 슬롯을 반환한다. 7. 외부 I/O(브로커 등)는 lock 밖에서 호출자가 수행한다(lock 보유 중 금지). -## Flow: `slot strategy-state 저장소` (BAT-042 / dcy-028, REQ-079~082) +## Flow: `slot strategy-state 저장소` (REQ-0009) -전략 데몬(예: trex-tsl)이 슬롯 점유 동안 누적하는 opaque 상태(예: TSL peak)를 슬롯 수명에 묶어 server-side 로 보관한다. core 는 `data` 를 해석하지 않는다(opaque UTF-8 + byte size cap 만 검증). +전략 상태는 슬롯 점유와 독립된 opaque 상태다. core는 inner `data`를 해석하지 않고 UTF-8 byte size cap만 검증하며, 하나의 슬롯별 map `data[strategy_instance_id] = {version,data,updated_at}`를 보관한다. -1. **read (piggyback, C-1)**: 데몬이 매 사이클 `trex.slots.req` 로 슬롯을 읽으면 각 행에 `occupancy_seq`(int)와, 보유 시 `strategy_state` 맵(`{: {version,data,updated_at}}`)이 함께 실린다. 맵은 핸들러가 반환 slot_id 들로 `StrategyStateStore.get_for_slots()` 1회 batched 조회 후 group by 해 조립한다(추가 왕복 없음). 슬롯 현재 occupancy_seq 와 다른 stale 엔트리는 노출하지 않는다. -2. **evaluate (pure)**: 데몬은 행의 `strategy_state[strategy_id]` 를 read-only 입력으로 `evaluate()` 에 넣는다(상태 write 는 evaluate 밖, 엔진이 수행). -3. **write (별도 RPC, C-2)**: 상태가 바뀌면 `trex.slots.strategy_state.set` 으로 `{slot_id, strategy_id, expected_version, expected_occupancy_seq, data}` 를 보낸다. core 는 슬롯의 현재 occupancy_seq·active 여부를 조회해 CAS 검증: 점유 불일치 → `occupancy_changed`, version 불일치 → `version_conflict`, inactive/미존재 → `slot_inactive`, size 초과 → `state_too_large`, 일치(없으면 expected_version=0) → commit + version+1. malformed → `bad_request`. 동일 입력 재전송은 무부작용. -4. **occupancy 가드 (C-2b, R-1)**: 슬롯은 monotonic `occupancy_seq` 를 갖고 새 점유 시작(`inactive→pending_buy`)마다 ++. 영구 슬롯 재사용 시 read v0→close→reopen→write v0 의 ABA(옛 점유 계산이 새 점유에 바인딩)를 occupancy_seq CAS 가 구조적으로 차단한다. -5. **close 삭제 (C-3)**: 슬롯이 어떤 경로로든 `→INACTIVE`(전략 매도/safety-sell/forced-close/operator) 되면 `transit_slot` 단일 전이 지점에서 해당 slot_id 의 모든 strategy_state 를 삭제한다. 재사용 슬롯은 새 occupancy_seq + 빈 상태로 시작 — 전략/데몬은 cleanup 책임이 없다(슬롯이 source of truth + GC). +1. **read (piggyback)**: `spy=false` 전략 요청은 required non-empty `strategy_instance_id`의 exact entry만 각 slot 행의 single `strategy_state` object로 받으며, 없으면 field가 생략된다. 다른 ID나 전체 map은 노출하지 않는다. `spy=true` 운영자 요청은 ID 없이도 각 slot의 complete instance map을 받고 watchdog을 갱신하지 않는다. +2. **write (별도 RPC)**: `trex.slots.strategy_state.set` payload는 `{slot_id, strategy_instance_id, expected_version, data}`다. active slot, size limit, 요청 entry version만 CAS한다. entry가 없으면 `expected_version=0`, 성공하면 그 entry version만 증가한다. `version_conflict`, `slot_inactive`, `state_too_large`, `bad_request:*`는 write를 적용하지 않는다. +3. **lifecycle/concurrency**: `→INACTIVE`와 재점유는 map을 삭제하거나 숨기지 않으며 version도 초기화하지 않는다. Slot 자체의 `occupancy_seq`는 기존대로 유지하지만 strategy state persistence/read/write에는 사용하지 않는다. 서로 다른 instance entry의 concurrent 성공 write는 모두 보존된다. -저장은 별도 `slot_strategy_state` 테이블(복합 PK `(slot_id, strategy_id)`), 슬롯 본체에는 `occupancy_seq` 컬럼 1개만 추가(full 3계층 분할은 후속 Discovery). full 3계층(occupancy 테이블 분할)·다중 writer 동시성·장기 보존은 범위 밖. +저장은 별도 `slot_strategy_state` 테이블의 slot-scoped JSON map이며 entry에는 identity metadata를 중복 저장하지 않는다. core-side instance registry, legacy mapping/backfill, caller 인증은 없다. ## Flow Constraints diff --git a/tests/adapters/persistence/test_postgresql_repository.py b/tests/adapters/persistence/test_postgresql_repository.py index 77bd96a..856cde1 100644 --- a/tests/adapters/persistence/test_postgresql_repository.py +++ b/tests/adapters/persistence/test_postgresql_repository.py @@ -155,8 +155,10 @@ def test_cancel_attempt_reservation_is_compare_and_set(pg_clean): assert repo.fetch_slot(decision.slot_id).cancel_state == "requested" -# dcy-028 / REQ-079·081 (BAT-042): strategy-state store + occupancy_seq column +# REQ-0009: slot-scoped strategy-instance state map. def test_strategy_state_store_pg(pg_clean): + from concurrent.futures import ThreadPoolExecutor + from trex_core.adapters.persistence.postgresql import PgStrategyStateStore from trex_core.application import strategy_state as ss @@ -169,31 +171,30 @@ def test_strategy_state_store_pg(pg_clean): reason="buy", actor="t") svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="filled", actor="t", qty=10, avg_price=100) - # occupancy_seq persisted - held = repo.fetch_slot(slot.slot_id) - assert held.occupancy_seq == 1 - - # first-create - r1 = store.set(slot.slot_id, "tsl_v1", expected_version=0, - expected_occupancy_seq=1, data="peak", current_occupancy_seq=1, - slot_active=True) + r1 = store.set(slot.slot_id, "instance-a", expected_version=0, data="peak", slot_active=True) assert r1.ok and r1.version == 1 - # version conflict - r2 = store.set(slot.slot_id, "tsl_v1", expected_version=0, - expected_occupancy_seq=1, data="stale", current_occupancy_seq=1, - slot_active=True) + r2 = store.set(slot.slot_id, "instance-a", expected_version=0, data="stale", slot_active=True) assert not r2.ok and r2.reason == ss.REASON_VERSION_CONFLICT - # batched read - got = store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) - assert got[slot.slot_id]["tsl_v1"].data == "peak" + missing = store.set(slot.slot_id, "instance-missing", expected_version=1, data="stale", slot_active=True) + assert not missing.ok and missing.reason == ss.REASON_VERSION_CONFLICT and missing.version == 0 + store.set(slot.slot_id, "instance-b", expected_version=0, data="other", slot_active=True) + got = store.get_for_slots([slot.slot_id]) + assert got[slot.slot_id]["instance-a"].data == "peak" + assert got[slot.slot_id]["instance-b"].data == "other" + + with ThreadPoolExecutor(max_workers=2) as executor: + writes = list(executor.map( + lambda item: store.set(slot.slot_id, item, expected_version=0, data=item, slot_active=True), + ("instance-c", "instance-d"), + )) + assert all(write.ok for write in writes) + assert {"instance-c", "instance-d"} <= set(store.get_for_slots([slot.slot_id])[slot.slot_id]) - # close deletes state (C-3) svc.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="sell", actor="t") svc.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, reason="sold", actor="t") - assert store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) == {} - # reuse bumps occupancy_seq + assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 1 reused = svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t") assert reused.occupancy_seq == 2 diff --git a/tests/application/test_ledger_option_d.py b/tests/application/test_ledger_option_d.py index e81dde2..8685773 100644 --- a/tests/application/test_ledger_option_d.py +++ b/tests/application/test_ledger_option_d.py @@ -107,7 +107,7 @@ def test_apply_cash_estimate_overlays_pending_delta(): repo = InMemorySlotRepository() repo.reconcile_universe("default", {"PF-1": "SYM"}) handler = make_slots_rpc_handler(repo, snapshot_store=store, ledger_outbox=outbox) - resp = handler(RpcRequest(payload={"filter": "all"}, correlation_id="test")) + resp = handler(RpcRequest(payload={"filter": "all", "strategy_instance_id": "test-instance"}, correlation_id="test")) slots = resp.payload["slots"] pf1 = next(s for s in slots if s["portfolio_id"] == "PF-1") assert pf1["cash_estimated"] is True diff --git a/tests/application/test_reconcile_warning_blocking.py b/tests/application/test_reconcile_warning_blocking.py index d1b348c..73487b7 100644 --- a/tests/application/test_reconcile_warning_blocking.py +++ b/tests/application/test_reconcile_warning_blocking.py @@ -693,7 +693,7 @@ def _wire_with_warn(warned_slot_ids=None): def _req(filter=None): - return RpcRequest(payload={"filter": filter or "all"}, correlation_id="t") + return RpcRequest(payload={"filter": filter or "all", "spy": True}, correlation_id="t") # REQ-104 AC-5: reconcile_warned=True for warned slot @@ -756,7 +756,7 @@ def test_slots_rpc_existing_warned_field_unchanged(): reason="filled", actor="test", avg_price=100.0, qty=10.0) warn_store = InMemoryReconcileWarningStore() handler = make_slots_rpc_handler(repo, snapshot_store=store, warn_store=warn_store) - resp = handler(RpcRequest(payload={"filter": "holding"}, correlation_id="t")) + resp = handler(RpcRequest(payload={"filter": "holding", "spy": True}, correlation_id="t")) row = resp.payload["slots"][0] # slot.qty=10 != hub asset_quantity=5 => warned=True (existing field) assert row["warned"] is True diff --git a/tests/application/test_slots_handler.py b/tests/application/test_slots_handler.py index 7493982..bea909f 100644 --- a/tests/application/test_slots_handler.py +++ b/tests/application/test_slots_handler.py @@ -43,6 +43,8 @@ def _req(filter=None, spy=False): payload["filter"] = filter if spy: payload["spy"] = True + else: + payload["strategy_instance_id"] = "test-instance" return RpcRequest(payload=payload, correlation_id="test-corr") diff --git a/tests/application/test_slots_handler_quantity_mismatch.py b/tests/application/test_slots_handler_quantity_mismatch.py index 582abc8..24d857f 100644 --- a/tests/application/test_slots_handler_quantity_mismatch.py +++ b/tests/application/test_slots_handler_quantity_mismatch.py @@ -74,6 +74,7 @@ def _make_handler(repo, warn_store, snapshot_items=None): def _req(filter=None): payload = {} + payload["spy"] = True if filter: payload["filter"] = filter return RpcRequest(payload=payload, correlation_id="test") diff --git a/tests/application/test_slots_handler_strategy.py b/tests/application/test_slots_handler_strategy.py index a278b3b..e6f7bac 100644 --- a/tests/application/test_slots_handler_strategy.py +++ b/tests/application/test_slots_handler_strategy.py @@ -67,6 +67,8 @@ def _req(filter=None, spy=False, strategy=None): payload["filter"] = filter if spy: payload["spy"] = True + else: + payload["strategy_instance_id"] = "test-instance" if strategy is not None: payload["strategy"] = strategy return RpcRequest(payload=payload, correlation_id="test-corr") diff --git a/tests/application/test_slots_handler_strategy_state.py b/tests/application/test_slots_handler_strategy_state.py index e717b14..8bb4e5a 100644 --- a/tests/application/test_slots_handler_strategy_state.py +++ b/tests/application/test_slots_handler_strategy_state.py @@ -1,93 +1,60 @@ -"""slots.req read piggyback of strategy_state + occupancy_seq (dcy-028 C-1). - -REQ-073 AC-7 / REQ-074 AC-6 / S-1: each slot row carries occupancy_seq and, -when present, a strategy_state map assembled from the store in one batched read. -strategy_* reasoning fields remain absent. -""" +"""Strategy and operator strategy-state visibility contracts for slots.req.""" from __future__ import annotations +import pytest + from trex_core.application.messaging import RpcRequest from trex_core.application.messaging_handlers import make_slots_rpc_handler from trex_core.application.strategy_state import InMemoryStrategyStateStore +from trex_core.application.watchdog import WatchdogStore from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository -INST = "default" - def _wire(): - repo = InMemorySlotRepository() - store = InMemoryStrategyStateStore() - service = SlotService(repo, instance_key=INST, strategy_state_store=store) - repo.reconcile_universe(INST, {"pf-0": "SYM"}) - slot = repo.list_slots(INST)[0] - service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, - reason="buy", actor="t") - service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, - reason="filled", actor="t", qty=10, avg_price=100) - handler = make_slots_rpc_handler(repo, instance_key=INST, - strategy_state_store=store) - return repo, store, service, slot, handler + repo, store, watchdog = InMemorySlotRepository(), InMemoryStrategyStateStore(), WatchdogStore() + service = SlotService(repo) + repo.reconcile_universe("default", {"pf": "SYM"}) + slot = repo.list_slots("default")[0] + service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t") + service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="fill", actor="t", qty=1, avg_price=1) + store.set(slot.slot_id, "instance-a", expected_version=0, data="a", slot_active=True) + store.set(slot.slot_id, "instance-b", expected_version=0, data="b", slot_active=True) + return slot, store, watchdog, make_slots_rpc_handler(repo, strategy_state_store=store, watchdog_store=watchdog) -def _rows(handler): - return handler(RpcRequest(payload={}, correlation_id="c")).payload["slots"] +def _call(handler, payload): + return handler(RpcRequest(payload=payload, correlation_id="c")).payload -def test_row_carries_occupancy_seq(): - _, _, _, slot, handler = _wire() - row = _rows(handler)[0] - assert row["occupancy_seq"] == 1 +def test_strategy_receives_only_exact_requested_entry(): + _, _, _, handler = _wire() + payload = _call(handler, {"strategy_instance_id": "instance-a"}) + state = payload["slots"][0]["strategy_state"] + assert state["data"] == "a" and set(state) == {"version", "data", "updated_at"} + assert "instance-b" not in repr(payload) -def test_row_carries_strategy_state_when_present(): - _, store, _, slot, handler = _wire() - store.set(slot.slot_id, "tsl_v1", expected_version=0, expected_occupancy_seq=1, - data='{"peak":100}', current_occupancy_seq=1, slot_active=True) - row = _rows(handler)[0] - assert "strategy_state" in row - assert row["strategy_state"]["tsl_v1"]["version"] == 1 - assert row["strategy_state"]["tsl_v1"]["data"] == '{"peak":100}' - assert "updated_at" in row["strategy_state"]["tsl_v1"] - - -def test_row_omits_strategy_state_when_absent(): - _, _, _, _, handler = _wire() - row = _rows(handler)[0] +def test_strategy_omits_absent_entry(): + _, _, _, handler = _wire() + row = _call(handler, {"strategy_instance_id": "absent"})["slots"][0] assert "strategy_state" not in row -def test_no_strategy_reasoning_fields(): - """REQ-074 AC-5 (REQ-116 additive 재도입 후에도 유지): slot-origin strategy fields absent. - - strategy_params / strategy_name / markets are now present (REQ-116 PortfolioItem-origin); - slot-origin strategy_id / sell_strategy_id / active_strategy_id remain absent. - """ - _, store, _, slot, handler = _wire() - store.set(slot.slot_id, "tsl_v1", expected_version=0, expected_occupancy_seq=1, - data="x", current_occupancy_seq=1, slot_active=True) - row = _rows(handler)[0] - # slot-origin fields must be absent (REQ-074 AC-5 unchanged) - for k in ("strategy_id", "sell_strategy_id", "active_strategy_id"): - assert k not in row - # PortfolioItem-origin fields now present (REQ-116 additive) - assert "strategy_params" in row # {} (safe default, no snapshot_store) - assert "strategy_name" in row # None - assert "markets" in row # [] +@pytest.mark.parametrize("identity", [None, "", False, 1, []]) +def test_malformed_strategy_request_has_exact_safe_envelope(identity): + _, _, watchdog, handler = _wire() + assert _call(handler, {"strategy_instance_id": identity}) == { + "slots": [], "available_slots": None, "degraded": False, + "reason": "bad_request:strategy_instance_id", + "message": "strategy_instance_id must be a non-empty string", + } + assert watchdog.last_seen("unused") is None -def test_reused_slot_shows_no_state(): - """After a full close + reuse cycle, no prior-occupancy state surfaces - (close-delete + read-side occupancy filter). occupancy_seq advances to 2.""" - repo, store, service, slot, handler = _wire() - store.set(slot.slot_id, "tsl_v1", expected_version=0, expected_occupancy_seq=1, - data="old", current_occupancy_seq=1, slot_active=True) - service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, - reason="sell", actor="t") - service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, - reason="sold", actor="t") - service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, - reason="buy", actor="t") # occupancy_seq -> 2 - row = _rows(handler)[0] - assert row["occupancy_seq"] == 2 - assert "strategy_state" not in row +def test_operator_without_or_with_id_receives_full_map_without_watchdog_update(): + slot, _, watchdog, handler = _wire() + for payload in ({"spy": True}, {"spy": True, "strategy_instance_id": "instance-a"}): + state = _call(handler, payload)["slots"][0]["strategy_state"] + assert set(state) == {"instance-a", "instance-b"} + assert watchdog.last_seen(str(slot.slot_id)) is None diff --git a/tests/application/test_strategy_state_set_handler.py b/tests/application/test_strategy_state_set_handler.py index 7cd794c..86bbe1e 100644 --- a/tests/application/test_strategy_state_set_handler.py +++ b/tests/application/test_strategy_state_set_handler.py @@ -1,10 +1,8 @@ -"""strategy_state.set write RPC handler tests (dcy-028 / REQ-080, BAT-042). - -Covers the full reject reason set + success + bad_request, with the handler -supplying the slot's current occupancy_seq / active-ness to the store CAS. -""" +"""strategy_state.set strategy-instance wire contract tests.""" from __future__ import annotations +import pytest + from trex_core.application import strategy_state as ss from trex_core.application.messaging import RpcRequest from trex_core.application.messaging_handlers import make_strategy_state_set_handler @@ -12,109 +10,41 @@ from trex_core.application.strategy_state import InMemoryStrategyStateStore from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository -INST = "default" - def _wire(): - repo = InMemorySlotRepository() - store = InMemoryStrategyStateStore() - service = SlotService(repo, instance_key=INST, strategy_state_store=store) - repo.reconcile_universe(INST, {"pf-0": "SYM"}) - slot = repo.list_slots(INST)[0] - # activate to holding (occupancy_seq -> 1) - service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, - reason="buy", actor="t") - service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, - reason="filled", actor="t", qty=10, avg_price=100) - handler = make_strategy_state_set_handler(repo, store) - return repo, store, service, slot, handler + repo, store = InMemorySlotRepository(), InMemoryStrategyStateStore() + service = SlotService(repo) + repo.reconcile_universe("default", {"pf": "SYM"}) + slot = repo.list_slots("default")[0] + service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t") + service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="fill", actor="t", qty=1, avg_price=1) + return repo, store, service, slot, make_strategy_state_set_handler(repo, store) -def _req(slot_id, *, ev=0, eo=1, data="peak", strat="tsl_v1"): - return RpcRequest(payload={ - "slot_id": str(slot_id), "strategy_id": strat, - "expected_version": ev, "expected_occupancy_seq": eo, "data": data, - }, correlation_id=f"{slot_id}:{strat}:{ev}") +def _req(slot_id, instance="instance-a", version=0, data="state"): + return RpcRequest(payload={"slot_id": str(slot_id), "strategy_instance_id": instance, "expected_version": version, "data": data}, correlation_id="same-correlation-id") -def test_first_write_ok(): +def test_write_uses_instance_identity_and_entry_cas(): _, _, _, slot, handler = _wire() - resp = handler(_req(slot.slot_id, ev=0)) - assert resp.payload["ok"] is True and resp.payload["version"] == 1 + assert handler(_req(slot.slot_id)).payload == {"ok": True, "slot_id": str(slot.slot_id), "strategy_instance_id": "instance-a", "version": 1} + conflict = handler(_req(slot.slot_id, version=0, data="stale")).payload + assert conflict["reason"] == ss.REASON_VERSION_CONFLICT and conflict["version"] == 1 -def test_version_conflict(): - _, _, _, slot, handler = _wire() - handler(_req(slot.slot_id, ev=0)) # version -> 1 - resp = handler(_req(slot.slot_id, ev=0, data="stale")) - assert resp.payload["ok"] is False - assert resp.payload["reason"] == ss.REASON_VERSION_CONFLICT - assert resp.payload["version"] == 1 +@pytest.mark.parametrize("identity", [None, "", 1, True, []]) +def test_malformed_instance_id_is_rejected_without_write(identity): + _, store, _, slot, handler = _wire() + response = handler(_req(slot.slot_id, identity)).payload + assert response["ok"] is False + assert response["reason"].startswith("bad_request:") + assert store.get_for_slots([slot.slot_id]) == {} -def test_occupancy_changed(): - _, _, _, slot, handler = _wire() - # daemon read occupancy 0 (before activation) but slot is occupancy 1 now - resp = handler(_req(slot.slot_id, ev=0, eo=0)) - assert resp.payload["ok"] is False - assert resp.payload["reason"] == ss.REASON_OCCUPANCY_CHANGED - assert resp.payload["occupancy_seq"] == 1 - - -def test_slot_inactive(): +def test_inactive_and_oversize_writes_do_not_change_existing_entry(): repo, store, service, slot, handler = _wire() - # close the slot - service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, - reason="sell", actor="t") - service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, - reason="sold", actor="t") - resp = handler(_req(slot.slot_id, ev=0, eo=1)) - assert resp.payload["ok"] is False - assert resp.payload["reason"] == ss.REASON_SLOT_INACTIVE - - -def test_unknown_slot_is_slot_inactive(): - from uuid import uuid4 - _, _, _, _, handler = _wire() - resp = handler(_req(uuid4(), ev=0, eo=1)) - assert resp.payload["ok"] is False - assert resp.payload["reason"] == ss.REASON_SLOT_INACTIVE - - -def test_state_too_large(): - _, _, _, slot, handler = _wire() - big = "a" * (ss._STATE_CAP_BYTES + 1) - resp = handler(_req(slot.slot_id, ev=0, data=big)) - assert resp.payload["ok"] is False - assert resp.payload["reason"] == ss.REASON_STATE_TOO_LARGE - - -def test_bad_request_missing_field(): - _, _, _, slot, handler = _wire() - resp = handler(RpcRequest(payload={"slot_id": str(slot.slot_id)}, - correlation_id="c")) - assert resp.payload["ok"] is False - assert resp.payload["reason"].startswith("bad_request:") - - -def test_bad_request_non_string_data(): - _, _, _, slot, handler = _wire() - resp = handler(RpcRequest(payload={ - "slot_id": str(slot.slot_id), "strategy_id": "tsl_v1", - "expected_version": 0, "expected_occupancy_seq": 1, "data": 123, - }, correlation_id="c")) - assert resp.payload["ok"] is False - assert resp.payload["reason"].startswith("bad_request:") - - -def test_reason_set_is_enumerated(): - """All non-ok reasons must be within the documented set (REQ-080 AC-6).""" - allowed_prefixes = { - ss.REASON_OCCUPANCY_CHANGED, ss.REASON_VERSION_CONFLICT, - ss.REASON_SLOT_INACTIVE, ss.REASON_STATE_TOO_LARGE, "bad_request", - } - _, _, _, slot, handler = _wire() - handler(_req(slot.slot_id, ev=0)) - bad = handler(_req(slot.slot_id, ev=0, data="x")) # version_conflict - reason = bad.payload["reason"].split(":")[0] - assert reason in allowed_prefixes + handler(_req(slot.slot_id)) + service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="sell", actor="t") + service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, reason="sold", actor="t") + assert handler(_req(slot.slot_id, version=1)).payload["reason"] == ss.REASON_SLOT_INACTIVE + assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 1 diff --git a/tests/application/test_strategy_state_store.py b/tests/application/test_strategy_state_store.py index 2b6963a..f76f742 100644 --- a/tests/application/test_strategy_state_store.py +++ b/tests/application/test_strategy_state_store.py @@ -1,118 +1,53 @@ -"""InMemoryStrategyStateStore CAS tests (dcy-028 / REQ-079, BAT-042). - -Covers the 6 CAS outcomes (occupancy+version commit / version_conflict / -occupancy_changed / slot_inactive / state_too_large / first-create), idempotent -re-send, batched read with stale-occupancy filtering, and delete-on-close. -""" +"""Slot-scoped strategy-instance state store contract tests.""" from __future__ import annotations +from concurrent.futures import ThreadPoolExecutor from uuid import uuid4 -import pytest - from trex_core.application import strategy_state as ss from trex_core.application.strategy_state import InMemoryStrategyStateStore -@pytest.fixture -def store(): - return InMemoryStrategyStateStore() - - -def _set(store, slot_id, *, ev=0, eo=1, data="x", cur_occ=1, active=True, strat="tsl_v1"): +def _set(store, slot_id, instance="instance-a", *, version=0, data="x", active=True): return store.set( - slot_id, strat, - expected_version=ev, expected_occupancy_seq=eo, data=data, - current_occupancy_seq=cur_occ, slot_active=active, + slot_id, instance, expected_version=version, data=data, slot_active=active ) -class TestCas: - def test_first_create(self, store): - sid = uuid4() - r = _set(store, sid, ev=0, eo=1, cur_occ=1, data='{"peak":100}') - assert r.ok and r.version == 1 - - def test_commit_increments_version(self, store): - sid = uuid4() - _set(store, sid, ev=0) - r = _set(store, sid, ev=1, data="next") - assert r.ok and r.version == 2 - - def test_version_conflict_does_not_store(self, store): - sid = uuid4() - _set(store, sid, ev=0, data="first") # version=1 - r = _set(store, sid, ev=0, data="stale") # expects 0, stored 1 - assert not r.ok and r.reason == ss.REASON_VERSION_CONFLICT and r.version == 1 - # original preserved - got = store.get_for_slots([sid], {sid: 1})[sid]["tsl_v1"] - assert got.data == "first" and got.version == 1 - - def test_occupancy_changed_does_not_store(self, store): - sid = uuid4() - # daemon read occupancy_seq=1 but slot is now occupancy 2 - r = _set(store, sid, ev=0, eo=1, cur_occ=2) - assert not r.ok and r.reason == ss.REASON_OCCUPANCY_CHANGED and r.occupancy_seq == 2 - assert store.get_for_slots([sid], {sid: 2}) == {} - - def test_slot_inactive_does_not_store(self, store): - sid = uuid4() - r = _set(store, sid, ev=0, active=False) - assert not r.ok and r.reason == ss.REASON_SLOT_INACTIVE - assert store.get_for_slots([sid], {sid: 1}) == {} - - def test_state_too_large(self, store, monkeypatch): - sid = uuid4() - big = "a" * (ss._STATE_CAP_BYTES + 1) - r = _set(store, sid, ev=0, data=big) - assert not r.ok and r.reason == ss.REASON_STATE_TOO_LARGE - - def test_idempotent_resend_is_conflict_not_double_apply(self, store): - sid = uuid4() - r1 = _set(store, sid, ev=0, data="v1") # -> version 1 - assert r1.ok and r1.version == 1 - # re-send identical expected_version=0 -> rejected (already advanced) - r2 = _set(store, sid, ev=0, data="v1") - assert not r2.ok and r2.reason == ss.REASON_VERSION_CONFLICT +def test_entry_versions_are_isolated(): + store, slot_id = InMemoryStrategyStateStore(), uuid4() + assert _set(store, slot_id, "a", data="a1").version == 1 + assert _set(store, slot_id, "b", data="b1").version == 1 + assert _set(store, slot_id, "a", version=1, data="a2").version == 2 + entries = store.get_for_slots([slot_id])[slot_id] + assert entries["a"].version == 2 and entries["a"].data == "a2" + assert entries["b"].version == 1 and entries["b"].data == "b1" + assert not hasattr(entries["a"], "strategy_id") + assert not hasattr(entries["a"], "occupancy_seq") -class TestAbaGuard: - def test_turnover_first_write_rejected(self, store): - """dcy-028 R-1 / S-4: read v0 of occupancy 1, slot reused as occupancy 2, - stale write against occupancy 1 must not bind to occupancy 2.""" - sid = uuid4() - # occupancy 2 is current; daemon computed against occupancy 1 (eo=1) - r = _set(store, sid, ev=0, eo=1, cur_occ=2, data="stale-peak") - assert not r.ok and r.reason == ss.REASON_OCCUPANCY_CHANGED - # new occupancy starts clean - assert store.get_for_slots([sid], {sid: 2}) == {} +def test_conflict_inactive_and_size_limit_preserve_entries(monkeypatch): + store, slot_id = InMemoryStrategyStateStore(), uuid4() + _set(store, slot_id, "a", data="saved") + assert _set(store, slot_id, "a", data="stale").reason == ss.REASON_VERSION_CONFLICT + assert _set(store, slot_id, "b", active=False).reason == ss.REASON_SLOT_INACTIVE + assert _set(store, slot_id, "b", data="x" * (ss._STATE_CAP_BYTES + 1)).reason == ss.REASON_STATE_TOO_LARGE + entries = store.get_for_slots([slot_id])[slot_id] + assert set(entries) == {"a"} and entries["a"].data == "saved" -class TestBatchedRead: - def test_get_for_slots_groups_by_slot_and_strategy(self, store): - a, b = uuid4(), uuid4() - _set(store, a, ev=0, cur_occ=1, strat="tsl_v1", data="A") - _set(store, a, ev=0, cur_occ=1, strat="buy_v1", data="A2") - _set(store, b, ev=0, cur_occ=1, strat="tsl_v1", data="B") - out = store.get_for_slots([a, b], {a: 1, b: 1}) - assert set(out[a]) == {"tsl_v1", "buy_v1"} - assert out[a]["tsl_v1"].data == "A" - assert out[b]["tsl_v1"].data == "B" - - def test_stale_occupancy_entries_invisible(self, store): - sid = uuid4() - _set(store, sid, ev=0, cur_occ=1, data="old") # stored at occupancy 1 - # slot is now occupancy 2 -> the occupancy-1 entry must not surface - assert store.get_for_slots([sid], {sid: 2}) == {} - - def test_empty_slot_ids(self, store): - assert store.get_for_slots([], {}) == {} +def test_concurrent_different_entries_both_survive(): + store, slot_id = InMemoryStrategyStateStore(), uuid4() + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(lambda key: _set(store, slot_id, key, data=key), ("a", "b"))) + assert all(result.ok for result in results) + assert set(store.get_for_slots([slot_id])[slot_id]) == {"a", "b"} -class TestDelete: - def test_delete_all_removes_every_strategy(self, store): - sid = uuid4() - _set(store, sid, ev=0, strat="tsl_v1") - _set(store, sid, ev=0, strat="buy_v1") - store.delete_all(sid) - assert store.get_for_slots([sid], {sid: 1}) == {} +def test_same_entry_competing_cas_has_one_winner(): + store, slot_id = InMemoryStrategyStateStore(), uuid4() + _set(store, slot_id, "a", data="first") + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(lambda data: _set(store, slot_id, "a", version=1, data=data), ("x", "y"))) + assert sum(result.ok for result in results) == 1 + assert store.get_for_slots([slot_id])[slot_id]["a"].version == 2 diff --git a/tests/domain/slot/test_occupancy_seq_and_close_delete.py b/tests/domain/slot/test_occupancy_seq_and_close_delete.py index 9be2b0f..24505cb 100644 --- a/tests/domain/slot/test_occupancy_seq_and_close_delete.py +++ b/tests/domain/slot/test_occupancy_seq_and_close_delete.py @@ -1,120 +1,23 @@ -"""Slot occupancy_seq bump + close-delete hook (dcy-028 / REQ-081, BAT-042). - -S-3: every close path (→INACTIVE) deletes the slot's strategy_state and a reused -slot starts with a new occupancy_seq + empty state. occupancy_seq is stable -across one holding episode and bumps on inactive→pending_buy. -""" +"""Slot occupancy remains intact while strategy state survives close/reopen.""" from __future__ import annotations from trex_core.application.strategy_state import InMemoryStrategyStateStore from trex_core.domain.slot import SlotService, SlotState from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository -INST = "default" - -def _seed(repo): - repo.reconcile_universe(INST, {"pf-0": "SYM"}) - return repo.list_slots(INST)[0] - - -def _service(repo, store=None): - return SlotService(repo, instance_key=INST, strategy_state_store=store) - - -def _activate_to_holding(service, slot_id): - service.transit_slot(slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, - reason="buy", actor="t") - return service.transit_slot(slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, - reason="filled", actor="t", qty=10, avg_price=100) - - -class TestOccupancySeq: - def test_bumps_on_inactive_to_pending_buy(self): - repo = InMemorySlotRepository() - service = _service(repo) - slot = _seed(repo) - assert slot.occupancy_seq == 0 - updated = service.transit_slot( - slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, - reason="buy", actor="t") - assert updated.occupancy_seq == 1 - - def test_stable_across_holding_episode(self): - repo = InMemorySlotRepository() - service = _service(repo) - slot = _seed(repo) - held = _activate_to_holding(service, slot.slot_id) - assert held.occupancy_seq == 1 - sold = service.transit_slot(slot.slot_id, SlotState.HOLDING, - SlotState.PENDING_SELL, reason="sell", actor="t") - assert sold.occupancy_seq == 1 # unchanged within the episode - - def test_increments_on_reuse(self): - repo = InMemorySlotRepository() - service = _service(repo) - slot = _seed(repo) - _activate_to_holding(service, slot.slot_id) - service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, - reason="sell", actor="t") - service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, - reason="sold", actor="t") - # reuse: occupancy_seq bumps from 1 to 2 - reused = service.transit_slot(slot.slot_id, SlotState.INACTIVE, - SlotState.PENDING_BUY, reason="buy", actor="t") - assert reused.occupancy_seq == 2 - - -class TestCloseDelete: - def _setup(self): - repo = InMemorySlotRepository() - store = InMemoryStrategyStateStore() - service = _service(repo, store) - slot = _seed(repo) - _activate_to_holding(service, slot.slot_id) - # daemon writes state during holding (occupancy_seq=1) - store.set(slot.slot_id, "tsl_v1", expected_version=0, - expected_occupancy_seq=1, data="peak", current_occupancy_seq=1, - slot_active=True) - assert store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) - return repo, store, service, slot - - def test_sell_fill_close_deletes_state(self): - repo, store, service, slot = self._setup() - service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, - reason="sell", actor="t") - service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, - reason="sold", actor="t") - assert store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) == {} - - def test_reused_slot_starts_empty_with_new_occupancy(self): - repo, store, service, slot = self._setup() - service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, - reason="sell", actor="t") - service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, - reason="sold", actor="t") - reused = service.transit_slot(slot.slot_id, SlotState.INACTIVE, - SlotState.PENDING_BUY, reason="buy", actor="t") - assert reused.occupancy_seq == 2 - assert store.get_for_slots([slot.slot_id], {slot.slot_id: 2}) == {} - - def test_error_close_path_also_deletes(self): - repo, store, service, slot = self._setup() - # holding → error → inactive (operator resolve) also clears state - service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.ERROR, - reason="err", actor="t") - service.transit_slot(slot.slot_id, SlotState.ERROR, SlotState.INACTIVE, - reason="resolve", actor="t") - assert store.get_for_slots([slot.slot_id], {slot.slot_id: 1}) == {} - - def test_no_store_injected_is_noop(self): - # backward-compat: service without store does not raise on close - repo = InMemorySlotRepository() - service = _service(repo, store=None) - slot = _seed(repo) - _activate_to_holding(service, slot.slot_id) - service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, - reason="sell", actor="t") - out = service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, - SlotState.INACTIVE, reason="sold", actor="t") - assert out.status is SlotState.INACTIVE +def test_close_preserves_state_and_reopen_continues_version(): + repo, store = InMemorySlotRepository(), InMemoryStrategyStateStore() + service = SlotService(repo, strategy_state_store=store) + repo.reconcile_universe("default", {"pf": "SYM"}) + slot = repo.list_slots("default")[0] + service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t") + service.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="fill", actor="t", qty=1, avg_price=1) + assert store.set(slot.slot_id, "instance", expected_version=0, data="old", slot_active=True).version == 1 + service.transit_slot(slot.slot_id, SlotState.HOLDING, SlotState.PENDING_SELL, reason="sell", actor="t") + inactive = service.transit_slot(slot.slot_id, SlotState.PENDING_SELL, SlotState.INACTIVE, reason="sold", actor="t") + assert inactive.occupancy_seq == 1 + assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance"].data == "old" + reopened = service.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t") + assert reopened.occupancy_seq == 2 + assert store.set(slot.slot_id, "instance", expected_version=1, data="new", slot_active=True).version == 2 diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 06bd717..7697179 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -28,7 +28,7 @@ from trex_core.application.ports import IntentRecord from trex_core.application.strategy_state import ( SetResult, StrategyStateEntry, - _cas, + _STATE_CAP_BYTES, ) from trex_core.domain.slot import Slot, SlotNotFoundError, SlotState, SlotTransitionRecord @@ -701,13 +701,7 @@ class PostgreSQLLedgerOutbox: class PgStrategyStateStore: - """slot_strategy_state-backed StrategyStateStore (dcy-028 / REQ-079). - - Stores opaque per-strategy state keyed by (slot_id, strategy_id). The CAS - decision is the same `_cas` used by the in-memory double; occupancy_seq and - slot_active are supplied by the caller (fetched from the slot) so the store - only owns the slot_strategy_state table. - """ + """PostgreSQL slot-scoped strategy-instance map store.""" def __init__(self, dsn: str, *, psycopg_module: Any = None) -> None: self._psycopg = _require_psycopg(psycopg_module) @@ -716,71 +710,65 @@ class PgStrategyStateStore: def set( self, slot_id: UUID, - strategy_id: str, + strategy_instance_id: str, *, - expected_occupancy_seq: int, expected_version: int, data: str, - current_occupancy_seq: int, slot_active: bool, ) -> SetResult: + if not slot_active: + return SetResult(ok=False, reason="slot_inactive") + if len(data.encode("utf-8")) > _STATE_CAP_BYTES: + return SetResult(ok=False, reason="state_too_large") + now = datetime.now(tz=timezone.utc) with self._psycopg.connect(self._dsn, autocommit=False) as conn: try: with conn.cursor() as cur: cur.execute( - "SELECT occupancy_seq, version, data, updated_at " - "FROM slot_strategy_state " - "WHERE slot_id = %s AND strategy_id = %s FOR UPDATE", - (slot_id, strategy_id), - ) - row = cur.fetchone() - existing = ( - StrategyStateEntry( - strategy_id=strategy_id, - occupancy_seq=int(row[0]), - version=int(row[1]), - data=row[2], - updated_at=row[3], - ) - if row is not None - else None - ) - result = _cas( - existing, - expected_occupancy_seq=expected_occupancy_seq, - expected_version=expected_version, - data=data, - current_occupancy_seq=current_occupancy_seq, - slot_active=slot_active, - ) - if not result.ok: - conn.rollback() - return result - cur.execute( - "INSERT INTO slot_strategy_state " - "(slot_id, strategy_id, occupancy_seq, version, data, updated_at) " - "VALUES (%s, %s, %s, %s, %s, %s) " - "ON CONFLICT (slot_id, strategy_id) DO UPDATE SET " - "occupancy_seq = EXCLUDED.occupancy_seq, " - "version = EXCLUDED.version, data = EXCLUDED.data, " - "updated_at = EXCLUDED.updated_at", + "INSERT INTO slot_strategy_state (slot_id, data) " + "SELECT %s, jsonb_build_object(%s, jsonb_build_object(" + "'version', 1, 'data', %s, 'updated_at', %s)) WHERE %s = 0 " + "ON CONFLICT (slot_id) DO UPDATE SET data = jsonb_set(" + "slot_strategy_state.data, ARRAY[%s], jsonb_build_object(" + "'version', %s + 1, 'data', %s, 'updated_at', %s), true) " + "WHERE COALESCE((slot_strategy_state.data -> %s ->> 'version')::bigint, 0) = %s " + "RETURNING data -> %s ->> 'version'", ( slot_id, - strategy_id, - current_occupancy_seq, - result.version, + strategy_instance_id, data, - datetime.now(tz=timezone.utc), + now, + expected_version, + strategy_instance_id, + expected_version, + data, + now, + strategy_instance_id, + expected_version, + strategy_instance_id, ), ) + row = cur.fetchone() + if row is None: + cur.execute( + "SELECT COALESCE((data -> %s ->> 'version')::bigint, 0) " + "FROM slot_strategy_state WHERE slot_id = %s", + (strategy_instance_id, slot_id), + ) + conflict = cur.fetchone() + return SetResult( + ok=False, + reason="version_conflict", + version=int(conflict[0]) if conflict is not None else 0, + ) conn.commit() - return result + return SetResult(ok=True, version=int(row[0])) except BaseException: conn.rollback() raise def get_for_slots( - self, slot_ids: list[UUID], current_occupancy_seq: dict[UUID, int] + self, slot_ids: list[UUID] ) -> dict[UUID, dict[str, StrategyStateEntry]]: if not slot_ids: return {} @@ -788,26 +776,16 @@ class PgStrategyStateStore: with self._psycopg.connect(self._dsn) as conn: with conn.cursor() as cur: cur.execute( - "SELECT slot_id, strategy_id, occupancy_seq, version, data, updated_at " + "SELECT slot_id, data " "FROM slot_strategy_state WHERE slot_id = ANY(%s)", (list(slot_ids),), ) for row in cur.fetchall(): sid = row[0] if isinstance(row[0], UUID) else UUID(str(row[0])) - # stale-occupancy entries are invisible (dcy-028 FR-3 defense) - if int(row[2]) != current_occupancy_seq.get(sid): - continue - out.setdefault(sid, {})[row[1]] = StrategyStateEntry( - strategy_id=row[1], - occupancy_seq=int(row[2]), - version=int(row[3]), - data=row[4], - updated_at=row[5], - ) + for instance_id, entry in row[1].items(): + out.setdefault(sid, {})[instance_id] = StrategyStateEntry( + version=int(entry["version"]), + data=entry["data"], + updated_at=datetime.fromisoformat(entry["updated_at"]), + ) return out - - def delete_all(self, slot_id: UUID) -> None: - with self._psycopg.connect(self._dsn, autocommit=False) as conn: - with conn.cursor() as cur: - cur.execute("DELETE FROM slot_strategy_state WHERE slot_id = %s", (slot_id,)) - conn.commit() diff --git a/trex_core/adapters/persistence/sql/027_replace_slot_strategy_state_with_instance_map.sql b/trex_core/adapters/persistence/sql/027_replace_slot_strategy_state_with_instance_map.sql new file mode 100644 index 0000000..0639e72 --- /dev/null +++ b/trex_core/adapters/persistence/sql/027_replace_slot_strategy_state_with_instance_map.sql @@ -0,0 +1,18 @@ +-- REQ-0009: state is one opaque map per slot, keyed by strategy_instance_id. +-- The old strategy_id/occupancy-scoped rows cannot be assigned an instance ID; +-- no mapping or backfill is permitted by this contract. +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'public' + AND table_name = 'slot_strategy_state' + AND column_name = 'strategy_id' + ) THEN + DROP TABLE slot_strategy_state; + END IF; +END $$; +CREATE TABLE slot_strategy_state ( + slot_id UUID PRIMARY KEY, + data JSONB NOT NULL DEFAULT '{}'::jsonb +); diff --git a/trex_core/application/messaging_handlers.py b/trex_core/application/messaging_handlers.py index 62c0281..99bb4e1 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -207,7 +207,7 @@ def make_slots_rpc_handler( ): """Build the RPC handler for `trex.slots.req` (REQ-073/074/116). - Request: {filter?: str, spy?: bool, strategy?: str} + Request: {filter?: str, spy?: bool, strategy?: str, strategy_instance_id?: str} filter: "all"|"inactive"|"pending"|"holding"|"error" (default "all") spy=true: watchdog not updated (operator view, REQ-034 CHG-01) strategy: exact-match on strategy_name (REQ-116, optional). When specified @@ -224,6 +224,15 @@ def make_slots_rpc_handler( body = req.payload or {} filter_val = str(body.get("filter") or "all").lower() spy = bool(body.get("spy", False)) + strategy_instance_id = body.get("strategy_instance_id") + if not spy and (not isinstance(strategy_instance_id, str) or not strategy_instance_id): + return RpcResponse(payload={ + "slots": [], + "available_slots": None, + "degraded": False, + "reason": "bad_request:strategy_instance_id", + "message": "strategy_instance_id must be a non-empty string", + }) # REQ-116 (BAT-055): optional strategy filter — exact-match on strategy_name. # None means no filter (all slots returned regardless of strategy_name). strategy_filter = _opt_str(body.get("strategy")) @@ -290,24 +299,29 @@ def make_slots_rpc_handler( row["qty"] = None row["qty_source"] = "unavailable" - # dcy-028 / REQ-073/074·079 (C-1): attach opaque strategy_state map per - # row via one batched store read (read piggyback — no extra round trip). - # Only entries matching the slot's current occupancy_seq are surfaced. + # Strategy requests receive only their exact entry; operator requests + # receive the complete slot map without changing watchdog behavior. if strategy_state_store is not None and slots: slot_ids = [s.slot_id for s in slots] - occ = {s.slot_id: getattr(s, "occupancy_seq", 0) for s in slots} - state_map = strategy_state_store.get_for_slots(slot_ids, occ) + state_map = strategy_state_store.get_for_slots(slot_ids) for row, s in zip(slot_rows, slots, strict=True): entries = state_map.get(s.slot_id) if entries: - row["strategy_state"] = { - strat: { - "version": e.version, - "data": e.data, - "updated_at": e.updated_at.isoformat(), + if spy: + row["strategy_state"] = { + instance_id: { + "version": e.version, + "data": e.data, + "updated_at": e.updated_at.isoformat(), + } + for instance_id, e in entries.items() + } + elif (entry := entries.get(strategy_instance_id)) is not None: + row["strategy_state"] = { + "version": entry.version, + "data": entry.data, + "updated_at": entry.updated_at.isoformat(), } - for strat, e in entries.items() - } # dcy-018 / REQ-029: cash-estimated overlay (best-effort) if ledger_outbox is not None: @@ -406,15 +420,12 @@ def make_strategy_state_set_handler( ): """Build the RPC handler for `trex.slots.strategy_state.set` (REQ-080). - Request: {slot_id, strategy_id, expected_version:int, - expected_occupancy_seq:int, data:str} - Response (ok): {ok:true, slot_id, strategy_id, version:int} - Response (fail): {ok:false, slot_id, strategy_id, reason, version?|occupancy_seq?} + Request: {slot_id, strategy_instance_id, expected_version:int, data:str} + Response (ok): {ok:true, slot_id, strategy_instance_id, version:int} + Response (fail): {ok:false, slot_id, strategy_instance_id, reason, version?} - reason ∈ {occupancy_changed, version_conflict, slot_inactive, - state_too_large, bad_request:}. The store owns the CAS - (REQ-079); the handler supplies the slot's current occupancy_seq and - active-ness and never interprets `data` (opaque). + The store owns entry-version CAS; the handler supplies active-ness and never + interprets opaque data. """ _ACTIVE_STATES = {"pending_buy", "holding", "pending_sell"} @@ -422,9 +433,10 @@ def make_strategy_state_set_handler( body = req.payload or {} try: slot_id = UUID(str(body["slot_id"])) - strategy_id = str(body["strategy_id"]) + strategy_instance_id = body["strategy_instance_id"] + if not isinstance(strategy_instance_id, str) or not strategy_instance_id: + raise ValueError("strategy_instance_id must be a non-empty string") expected_version = int(body["expected_version"]) - expected_occupancy_seq = int(body["expected_occupancy_seq"]) data = body["data"] if not isinstance(data, str): raise ValueError("data must be a string") @@ -432,27 +444,23 @@ def make_strategy_state_set_handler( return RpcResponse(payload={ "ok": False, "slot_id": body.get("slot_id"), - "strategy_id": body.get("strategy_id"), + "strategy_instance_id": body.get("strategy_instance_id"), "reason": f"bad_request:{exc}", }) slot = slot_repo.fetch_slot(slot_id) slot_active = slot is not None and slot.status.value in _ACTIVE_STATES - current_occupancy_seq = getattr(slot, "occupancy_seq", 0) if slot else 0 - result = strategy_state_store.set( slot_id, - strategy_id, - expected_occupancy_seq=expected_occupancy_seq, + strategy_instance_id, expected_version=expected_version, data=data, - current_occupancy_seq=current_occupancy_seq, slot_active=slot_active, ) payload = { "ok": result.ok, "slot_id": str(slot_id), - "strategy_id": strategy_id, + "strategy_instance_id": strategy_instance_id, } if result.ok: payload["version"] = result.version @@ -460,8 +468,6 @@ def make_strategy_state_set_handler( payload["reason"] = result.reason if result.version is not None: payload["version"] = result.version - if result.occupancy_seq is not None: - payload["occupancy_seq"] = result.occupancy_seq return RpcResponse(payload=payload) return handler diff --git a/trex_core/application/portfolio.py b/trex_core/application/portfolio.py index f9bfbf6..abe519b 100644 --- a/trex_core/application/portfolio.py +++ b/trex_core/application/portfolio.py @@ -564,7 +564,7 @@ def slot_to_dict( "forced_close_at": s.forced_close_at.isoformat() if s.forced_close_at else None, "universe_detached": getattr(s, "retired_at", None) is not None, # dcy-028/REQ-081 (BAT-042): per-occupancy guard counter — the strategy - # daemon echoes this as expected_occupancy_seq on strategy_state writes. + # Slot occupancy remains domain state; strategy-state writes do not use it. "occupancy_seq": getattr(s, "occupancy_seq", 0), } @@ -638,4 +638,3 @@ def _sanitize(message: str) -> str: break return out[:200] - diff --git a/trex_core/application/strategy_state.py b/trex_core/application/strategy_state.py index b782665..78d638b 100644 --- a/trex_core/application/strategy_state.py +++ b/trex_core/application/strategy_state.py @@ -1,27 +1,4 @@ -"""Slot strategy-state store (dcy-028 / REQ-079, BAT-042). - -A strategy daemon (e.g. trex-tsl) accumulates opaque state across a slot's -occupancy — for a Trailing Stop Loss the peak price since entry. `evaluate()` -must stay pure, so the state cannot live in the daemon (lost on restart) nor in -an external store (orphan-cleanup problem). Since the state's lifecycle matches -the slot occupancy exactly, trex-core stores it server-side keyed by -`(slot_id, strategy_id)`, round-trips it on the slots snapshot (read), accepts -version-CAS writes via a dedicated RPC, and deletes it when the slot closes. - -trex-core never interprets `data` (opaque UTF-8 string). It validates only -(1) valid UTF-8 and (2) a byte size cap — encoding (JSON/base64) is the -strategy's business (dcy-028 OQ-1). - -Occupancy guard (dcy-028 R-1 / REQ-081): a permanent slot is reused, so two -occupancies of the same slot both start at version 0. A write computed against -occupancy X must not land on a later occupancy Y. The slot carries a monotonic -`occupancy_seq`; the CAS rejects a write whose `expected_occupancy_seq` differs -from the slot's current value (`occupancy_changed`). The version CAS alone -cannot catch this at the first write (0 vs 0). - -Single-writer assumption: one daemon operates a given slot·strategy. The CAS is -for retry / duplicate / occupancy-turnover safety, not multi-writer concurrency. -""" +"""Slot-scoped opaque strategy-instance state.""" from __future__ import annotations import os @@ -31,13 +8,8 @@ from datetime import datetime, timezone from typing import Optional, Protocol from uuid import UUID -# dcy-028 OQ-1: hard ceiling for the opaque `data` blob, measured in UTF-8 -# bytes. Operators target 1~2KB for the known TSL use case; 4KB is the cap. _STATE_CAP_BYTES = int(os.environ.get("TREX_STRATEGY_STATE_CAP_BYTES", "4096")) -# Reject reason tokens (dcy-028 / REQ-080). The full enumerated set; consumers -# fail-closed on anything unrecognized. -REASON_OCCUPANCY_CHANGED = "occupancy_changed" REASON_VERSION_CONFLICT = "version_conflict" REASON_SLOT_INACTIVE = "slot_inactive" REASON_STATE_TOO_LARGE = "state_too_large" @@ -45,10 +17,8 @@ REASON_STATE_TOO_LARGE = "state_too_large" @dataclass(frozen=True) class StrategyStateEntry: - """One stored opaque state value for a (slot_id, strategy_id).""" + """One opaque entry in a slot's strategy_instance_id map.""" - strategy_id: str - occupancy_seq: int version: int data: str updated_at: datetime @@ -56,64 +26,20 @@ class StrategyStateEntry: @dataclass(frozen=True) class SetResult: - """Outcome of a CAS set. ok=True carries the new version; else `reason`.""" - ok: bool version: Optional[int] = None reason: Optional[str] = None - occupancy_seq: Optional[int] = None # current slot occupancy_seq on conflict def _byte_len(data: str) -> int: return len(data.encode("utf-8")) -class StrategyStateStore(Protocol): - def set( - self, - slot_id: UUID, - strategy_id: str, - *, - expected_occupancy_seq: int, - expected_version: int, - data: str, - current_occupancy_seq: int, - slot_active: bool, - ) -> SetResult: - """occupancy_seq + version CAS upsert of opaque `data` (dcy-028 FR-2).""" - ... - - def get_for_slots( - self, slot_ids: list[UUID], current_occupancy_seq: dict[UUID, int] - ) -> dict[UUID, dict[str, StrategyStateEntry]]: - """Batched read for the snapshot (C-1). Returns slot_id -> {strategy_id: entry}. - - Only entries whose stored occupancy_seq matches the slot's current - occupancy_seq are returned (stale entries from a prior occupancy are - invisible — defense in depth alongside close-delete, dcy-028 FR-3).""" - ... - - def delete_all(self, slot_id: UUID) -> None: - """Delete all strategy_state for a slot (close hook, dcy-028 C-3).""" - ... - - def _cas( - existing: Optional[StrategyStateEntry], - *, - expected_occupancy_seq: int, - expected_version: int, - data: str, - current_occupancy_seq: int, - slot_active: bool, + existing: Optional[StrategyStateEntry], *, expected_version: int, data: str, slot_active: bool ) -> SetResult: - """Pure CAS decision shared by all store implementations (dcy-028 FR-2).""" if not slot_active: return SetResult(ok=False, reason=REASON_SLOT_INACTIVE) - if expected_occupancy_seq != current_occupancy_seq: - return SetResult( - ok=False, reason=REASON_OCCUPANCY_CHANGED, occupancy_seq=current_occupancy_seq - ) if _byte_len(data) > _STATE_CAP_BYTES: return SetResult(ok=False, reason=REASON_STATE_TOO_LARGE) stored_version = existing.version if existing is not None else 0 @@ -122,62 +48,54 @@ def _cas( return SetResult(ok=True, version=stored_version + 1) +class StrategyStateStore(Protocol): + def set( + self, + slot_id: UUID, + strategy_instance_id: str, + *, + expected_version: int, + data: str, + slot_active: bool, + ) -> SetResult: ... + + def get_for_slots(self, slot_ids: list[UUID]) -> dict[UUID, dict[str, StrategyStateEntry]]: ... + + class InMemoryStrategyStateStore: - """Process-local strategy-state store (paper / tests / explicit double).""" + """Process-local slot-scoped state-map store.""" def __init__(self) -> None: - # (slot_id, strategy_id) -> StrategyStateEntry - self._store: dict[tuple[UUID, str], StrategyStateEntry] = {} + self._store: dict[UUID, dict[str, StrategyStateEntry]] = {} self._lock = threading.Lock() def set( self, slot_id: UUID, - strategy_id: str, + strategy_instance_id: str, *, - expected_occupancy_seq: int, expected_version: int, data: str, - current_occupancy_seq: int, slot_active: bool, ) -> SetResult: with self._lock: - existing = self._store.get((slot_id, strategy_id)) + entries = self._store.setdefault(slot_id, {}) + existing = entries.get(strategy_instance_id) result = _cas( - existing, - expected_occupancy_seq=expected_occupancy_seq, - expected_version=expected_version, - data=data, - current_occupancy_seq=current_occupancy_seq, - slot_active=slot_active, + existing, expected_version=expected_version, data=data, slot_active=slot_active ) if not result.ok: + if not entries: + self._store.pop(slot_id, None) return result - self._store[(slot_id, strategy_id)] = StrategyStateEntry( - strategy_id=strategy_id, - occupancy_seq=current_occupancy_seq, + entries[strategy_instance_id] = StrategyStateEntry( version=result.version, # type: ignore[arg-type] data=data, updated_at=datetime.now(tz=timezone.utc), ) return result - def get_for_slots( - self, slot_ids: list[UUID], current_occupancy_seq: dict[UUID, int] - ) -> dict[UUID, dict[str, StrategyStateEntry]]: + def get_for_slots(self, slot_ids: list[UUID]) -> dict[UUID, dict[str, StrategyStateEntry]]: wanted = set(slot_ids) - out: dict[UUID, dict[str, StrategyStateEntry]] = {} with self._lock: - for (sid, strat), entry in self._store.items(): - if sid not in wanted: - continue - # stale-occupancy entries are invisible (dcy-028 FR-3 defense) - if entry.occupancy_seq != current_occupancy_seq.get(sid): - continue - out.setdefault(sid, {})[strat] = entry - return out - - def delete_all(self, slot_id: UUID) -> None: - with self._lock: - for key in [k for k in self._store if k[0] == slot_id]: - del self._store[key] + return {sid: dict(entries) for sid, entries in self._store.items() if sid in wanted} diff --git a/trex_core/domain/slot/service.py b/trex_core/domain/slot/service.py index d86e30f..b470401 100644 --- a/trex_core/domain/slot/service.py +++ b/trex_core/domain/slot/service.py @@ -52,8 +52,6 @@ class SlotService: # does not serialize transitions across instances (NFR-1). self._instance_key = instance_key self._lock_id = slot_advisory_lock_id(instance_key) - # dcy-028/REQ-081 (C-3): when injected, all strategy_state for a slot is - # deleted on close (→INACTIVE). None keeps legacy behaviour (no store). self._strategy_state_store = strategy_state_store @property @@ -155,11 +153,6 @@ class SlotService: extra=extra or {}, ) ) - # dcy-028/REQ-081 (C-3): close deletes the slot's strategy_state so a - # reused slot starts empty. Covers every close path (strategy sell, - # safety-sell, forced-close, operator) since all funnel to →INACTIVE. - if to_state is SlotState.INACTIVE and self._strategy_state_store is not None: - self._strategy_state_store.delete_all(slot_id) return updated @staticmethod