diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py index 9a2a312..3a35cd3 100644 --- a/tests/adapters/persistence/test_migration_and_drift.py +++ b/tests/adapters/persistence/test_migration_and_drift.py @@ -28,6 +28,7 @@ _EXPECTED_TABLES = { "broker_account_snapshot", # dcy-041 / GitHub #45 "ledger_skip", # GitHub #68 skipped-ledger durable audit "reconcile_notification_state", # req-0007 / ac-0024 + "slot_strategy_state", # req-0009 / ac-0027 } @@ -59,6 +60,25 @@ def test_daily_slot_columns_match_expected_contract(pg_clean): assert actual == set(EXPECTED_SLOT_FIELDS) +def test_strategy_state_schema_matches_instance_map_contract(pg_clean): + with psycopg.connect(pg_clean) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT column_name, data_type, udt_name " + "FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = 'slot_strategy_state'" + ) + columns = {row[0]: (row[1], row[2]) for row in cur.fetchall()} + cur.execute( + "SELECT a.attname FROM pg_index i " + "JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey) " + "WHERE i.indrelid = 'slot_strategy_state'::regclass AND i.indisprimary" + ) + primary_key = {row[0] for row in cur.fetchall()} + assert columns == {"slot_id": ("uuid", "uuid"), "data": ("jsonb", "jsonb")} + assert primary_key == {"slot_id"} + + def test_drift_detected_when_contract_extended(pg_clean): """Negative case: a contract field absent from the table is caught.""" actual = _columns(pg_clean, "daily_slot") diff --git a/tests/adapters/persistence/test_postgresql_repository.py b/tests/adapters/persistence/test_postgresql_repository.py index 856cde1..c422c00 100644 --- a/tests/adapters/persistence/test_postgresql_repository.py +++ b/tests/adapters/persistence/test_postgresql_repository.py @@ -164,7 +164,7 @@ def test_strategy_state_store_pg(pg_clean): repo = PostgreSQLSlotRepository(pg_clean) store = PgStrategyStateStore(pg_clean) - svc = SlotService(repo, instance_key=INST, strategy_state_store=store) + svc = SlotService(repo, instance_key=INST) repo.reconcile_universe(INST, {"pf-a": "005930"}) slot = repo.find_slot_by_portfolio(INST, "pf-a") svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, @@ -189,12 +189,58 @@ def test_strategy_state_store_pg(pg_clean): )) assert all(write.ok for write in writes) assert {"instance-c", "instance-d"} <= set(store.get_for_slots([slot.slot_id])[slot.slot_id]) + assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 1 + + # Separate connections race on the same entry/version: only one CAS wins. + with ThreadPoolExecutor(max_workers=2) as executor: + competing = list(executor.map( + lambda data: store.set( + slot.slot_id, "instance-a", expected_version=1, data=data, slot_active=True, + ), + ("winner-one", "winner-two"), + )) + assert sum(result.ok for result in competing) == 1, competing + assert sum(result.reason == ss.REASON_VERSION_CONFLICT for result in competing) == 1 + winner = next(data for data, result in zip(("winner-one", "winner-two"), competing) if result.ok) + assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].data == winner + assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 2 + + # A fresh adapter sees the byte-identical persisted entry and can continue CAS. + before_restart = store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"] + after_restart = PgStrategyStateStore(pg_clean).get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"] + assert after_restart == before_restart 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]["instance-a"].version == 1 + assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"].version == 2 reused = svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, - reason="buy", actor="t") + reason="buy", actor="t") assert reused.occupancy_seq == 2 + rehydrated = PgStrategyStateStore(pg_clean) + persisted = rehydrated.get_for_slots([slot.slot_id])[slot.slot_id]["instance-a"] + assert persisted == before_restart + assert rehydrated.set( + slot.slot_id, "instance-a", expected_version=2, data="continued", slot_active=True, + ).version == 3 + + +def test_strategy_state_write_cannot_follow_close(pg_clean): + """The store's locked lifecycle read rejects a write after close wins.""" + from trex_core.adapters.persistence.postgresql import PgStrategyStateStore + + repo = PostgreSQLSlotRepository(pg_clean) + svc = SlotService(repo, instance_key=INST) + store = PgStrategyStateStore(pg_clean) + repo.reconcile_universe(INST, {"pf-close": "005930"}) + slot = repo.find_slot_by_portfolio(INST, "pf-close") + svc.transit_slot(slot.slot_id, SlotState.INACTIVE, SlotState.PENDING_BUY, reason="buy", actor="t") + svc.transit_slot(slot.slot_id, SlotState.PENDING_BUY, SlotState.HOLDING, reason="fill", actor="t", qty=1) + assert store.set(slot.slot_id, "instance", expected_version=0, data="before", slot_active=True).ok + + 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") + result = store.set(slot.slot_id, "instance", expected_version=1, data="after", slot_active=True) + assert not result.ok and result.reason == "slot_inactive" + assert store.get_for_slots([slot.slot_id])[slot.slot_id]["instance"].data == "before" diff --git a/tests/application/test_slots_handler_quantity_mismatch.py b/tests/application/test_slots_handler_quantity_mismatch.py index 24d857f..124a495 100644 --- a/tests/application/test_slots_handler_quantity_mismatch.py +++ b/tests/application/test_slots_handler_quantity_mismatch.py @@ -332,6 +332,33 @@ def test_ac7_warn_store_failure_fail_closed(): assert "excluded_slots" in resp.payload +def test_ac23_exclusion_precedes_broker_and_strategy_state_readers(): + repo = InMemorySlotRepository() + svc = SlotService(repo) + warn_store = InMemoryReconcileWarningStore() + slot = _make_holding_slot(repo, svc, portfolio_id="pf-A", symbol=SYM_A, qty=10) + warn_store.set_warned(str(slot.slot_id), portfolio_id="pf-A", case_code="C") + + class ExplodingReader: + def holdings_for_slots_rpc(self): + raise AssertionError("excluded slot reached broker reader") + + class ExplodingStateStore: + def get_for_slots(self, _slot_ids): + raise AssertionError("excluded slot reached state reader") + + handler = make_slots_rpc_handler( + repo, + instance_key=INSTANCE, + warn_store=warn_store, + broker_quantity_reader=ExplodingReader(), + strategy_state_store=ExplodingStateStore(), + ) + response = handler(_req(filter="holding")).payload + assert response["slots"] == [] + assert response["excluded_slots"] == {"reconcile_quantity_mismatch": 1} + + # --------------------------------------------------------------------------- # AC-8: intent rejected if C-case warning set after last slots RPC (race) # --------------------------------------------------------------------------- diff --git a/tests/application/test_slots_handler_strategy_state.py b/tests/application/test_slots_handler_strategy_state.py index 8bb4e5a..cde4cc9 100644 --- a/tests/application/test_slots_handler_strategy_state.py +++ b/tests/application/test_slots_handler_strategy_state.py @@ -58,3 +58,14 @@ def test_operator_without_or_with_id_receives_full_map_without_watchdog_update() 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 + + +@pytest.mark.parametrize("spy", ["false", "true", 1, [], {}]) +def test_non_boolean_spy_is_rejected_without_visibility_or_watchdog_update(spy): + slot, _, watchdog, handler = _wire() + payload = _call(handler, {"spy": spy}) + assert payload == { + "slots": [], "available_slots": None, "degraded": False, + "reason": "bad_request:spy", "message": "spy must be a boolean", + } + assert watchdog.last_seen(str(slot.slot_id)) is None diff --git a/tests/conftest.py b/tests/conftest.py index 264e656..fc525ef 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -27,6 +27,7 @@ _PG_TABLES = ( "daily_symbol_entries", "daily_slot", "reconcile_notification_state", + "slot_strategy_state", ) 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 24505cb..ac48dda 100644 --- a/tests/domain/slot/test_occupancy_seq_and_close_delete.py +++ b/tests/domain/slot/test_occupancy_seq_and_close_delete.py @@ -8,7 +8,7 @@ from trex_core.domain.slot.in_memory_repository import InMemorySlotRepository def test_close_preserves_state_and_reopen_continues_version(): repo, store = InMemorySlotRepository(), InMemoryStrategyStateStore() - service = SlotService(repo, strategy_state_store=store) + 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") diff --git a/trex_core/adapters/persistence/postgresql.py b/trex_core/adapters/persistence/postgresql.py index 7697179..cbe5154 100644 --- a/trex_core/adapters/persistence/postgresql.py +++ b/trex_core/adapters/persistence/postgresql.py @@ -716,53 +716,53 @@ class PgStrategyStateStore: data: str, 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: + # Lock the lifecycle row before changing state. A close UPDATE + # and this write therefore have one serial order: a writer that + # follows a close observes INACTIVE and cannot recreate state. cur.execute( - "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_instance_id, - data, - now, - expected_version, - strategy_instance_id, - expected_version, - data, - now, - strategy_instance_id, - expected_version, - strategy_instance_id, - ), + "SELECT status FROM daily_slot WHERE slot_id = %s::uuid FOR UPDATE", + (slot_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() + slot = cur.fetchone() + if slot is None or slot[0] not in {"pending_buy", "holding", "pending_sell"}: + return SetResult(ok=False, reason="slot_inactive") + cur.execute( + "SELECT data FROM slot_strategy_state WHERE slot_id = %s::uuid", + (slot_id,), + ) + state_row = cur.fetchone() + state_map = state_row[0] if state_row is not None else {} + entry = state_map.get(strategy_instance_id) + version = int(entry["version"]) if entry is not None else 0 + if expected_version != version: return SetResult( ok=False, reason="version_conflict", - version=int(conflict[0]) if conflict is not None else 0, + version=version, + ) + entry_json = "jsonb_build_object('version', %s::bigint, " \ + "'data', %s::text, 'updated_at', %s::timestamptz)" + if state_row is None: + cur.execute( + "INSERT INTO slot_strategy_state (slot_id, data) VALUES (" + "%s::uuid, jsonb_build_object(%s::text, " + entry_json + "))", + (slot_id, strategy_instance_id, version + 1, data, now), + ) + else: + cur.execute( + "UPDATE slot_strategy_state SET data = jsonb_set(data, " + "ARRAY[%s::text]::text[], " + entry_json + ", true) " + "WHERE slot_id = %s::uuid", + (strategy_instance_id, version + 1, data, now, slot_id), ) conn.commit() - return SetResult(ok=True, version=int(row[0])) + return SetResult(ok=True, version=version + 1) except BaseException: conn.rollback() raise 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 index 0639e72..b875a8d 100644 --- 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 @@ -12,7 +12,7 @@ BEGIN DROP TABLE slot_strategy_state; END IF; END $$; -CREATE TABLE slot_strategy_state ( +CREATE TABLE IF NOT EXISTS 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 99bb4e1..9e5f291 100644 --- a/trex_core/application/messaging_handlers.py +++ b/trex_core/application/messaging_handlers.py @@ -223,7 +223,16 @@ def make_slots_rpc_handler( def handler(req: RpcRequest) -> RpcResponse: body = req.payload or {} filter_val = str(body.get("filter") or "all").lower() - spy = bool(body.get("spy", False)) + spy_value = body.get("spy", False) + if not isinstance(spy_value, bool): + return RpcResponse(payload={ + "slots": [], + "available_slots": None, + "degraded": False, + "reason": "bad_request:spy", + "message": "spy must be a boolean", + }) + spy = spy_value 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={ @@ -268,10 +277,42 @@ def make_slots_rpc_handler( for s in slots ] + # AC-0023 exclusion must precede all optional data readers so excluded + # slots cannot expose state or make an otherwise valid response fail. + excluded_count = 0 + if warn_store is not None: + try: + c_warned_ids = warn_store.list_c_case_warned() + except Exception: # noqa: BLE001 + _log.exception("warn_store.list_c_case_warned() failed — fail-closed") + return RpcResponse(payload={ + "slots": [], + "available_slots": None, + "degraded": True, + "reason": "warn_store_unavailable", + "message": ( + "reconcile warning store unavailable; " + "returning no slots (fail-closed)" + ), + "excluded_slots": {"reconcile_quantity_mismatch": 0}, + }) + if c_warned_ids: + included = [ + (slot, row) + for slot, row in zip(slots, slot_rows, strict=True) + if not ( + slot.status.value == "holding" + and str(slot.slot_id) in c_warned_ids + ) + ] + excluded_count = len(slots) - len(included) + slots = [slot for slot, _ in included] + slot_rows = [row for _, row in included] + broker_qty_by_symbol: dict[str, int] = {} broker_qty_available = False broker_qty_captured_at: Optional[datetime] = None - if broker_quantity_reader is not None: + if broker_quantity_reader is not None and slots: try: holdings, broker_qty_captured_at = broker_quantity_reader.holdings_for_slots_rpc() broker_qty_by_symbol = { @@ -361,43 +402,6 @@ def make_slots_rpc_handler( if strategy_filter is not None: slot_rows = [r for r in slot_rows if r.get("strategy_name") == strategy_filter] - # AC-0023: exclude holding slots with active case_code='C' reconcile warning. - # Fail-closed: if warn_store exists but throws, return degraded=True / no slots. - excluded_count = 0 - if warn_store is not None: - try: - c_warned_ids = warn_store.list_c_case_warned() - except Exception: # noqa: BLE001 - _log.exception("warn_store.list_c_case_warned() failed — fail-closed") - return RpcResponse(payload={ - "slots": [], - "available_slots": available_slots, - "degraded": True, - "reason": "warn_store_unavailable", - "message": ( - "reconcile warning store unavailable; " - "returning no slots (fail-closed)" - ), - "excluded_slots": {"reconcile_quantity_mismatch": 0}, - }) - if c_warned_ids: - # Rebuild slot lookup from the original `slots` list (before strategy - # filter reduced slot_rows). slot_rows[i] corresponds to slots[i] - # only before filtering; use slot_id stored in each row. - slot_by_id = {str(s.slot_id): s for s in slots} - filtered_rows: list[dict] = [] - for row in slot_rows: - slot_obj = slot_by_id.get(str(row.get("slot_id", ""))) - if ( - slot_obj is not None - and str(slot_obj.slot_id) in c_warned_ids - and slot_obj.status.value == "holding" - ): - excluded_count += 1 - else: - filtered_rows.append(row) - slot_rows = filtered_rows - base_payload: dict = { "slots": slot_rows, "available_slots": available_slots, diff --git a/trex_core/domain/slot/service.py b/trex_core/domain/slot/service.py index b470401..29c6789 100644 --- a/trex_core/domain/slot/service.py +++ b/trex_core/domain/slot/service.py @@ -44,15 +44,12 @@ class SlotService: self, repository: SlotRepository, instance_key: str = "default", - *, - strategy_state_store: Any = None, ) -> None: self._repo = repository # dcy-020/REQ-057: advisory lock is scoped per instance so a shared DB # does not serialize transitions across instances (NFR-1). self._instance_key = instance_key self._lock_id = slot_advisory_lock_id(instance_key) - self._strategy_state_store = strategy_state_store @property def instance_key(self) -> str: diff --git a/trex_core/runtime/daemon.py b/trex_core/runtime/daemon.py index f1f68ea..9e8e61e 100644 --- a/trex_core/runtime/daemon.py +++ b/trex_core/runtime/daemon.py @@ -182,13 +182,12 @@ class Daemon: # Persistence + services self.slot_repo = PostgreSQLSlotRepository(config.dsn) self.intent_store = PostgreSQLIntentStore(config.dsn) - # dcy-028 / REQ-079·081 (BAT-042): opaque strategy-state store. Injected - # into the slot service so close (→INACTIVE) deletes the slot's state. + # dcy-028 / REQ-079·081 (BAT-042): opaque strategy-state store persists + # independently of slot occupancy and is shared by both RPC handlers. self.strategy_state_store = PgStrategyStateStore(config.dsn) self.slot_service = SlotService( self.slot_repo, instance_key=config.instance_key, - strategy_state_store=self.strategy_state_store, ) self.outbox = PostgreSQLLedgerOutbox(config.dsn) @@ -458,7 +457,7 @@ class Daemon: topology.slots_queue(self.config.instance_key), handler, self.stop) def _strategy_state_set_loop(self) -> None: - # dcy-028 / REQ-080: opaque strategy-state write RPC (occupancy_seq+version CAS). + # dcy-028 / REQ-080: opaque strategy-state write RPC (entry-version CAS). transport = _build_transport(self.config) self._transports.append(transport) try: