Merge pull request #36 from fourblocks2020/feat/slot-filled-no-route

fix(slot-filled): 매도 체결 이벤트 NO_ROUTE 라우팅 갭 해소 (M3)

tests/application/test_messaging_event_publisher.py

@@ -9,7 +9,7 @@ from trex_core.adapters.messaging import topology
 from trex_core.adapters.messaging.in_memory import InMemoryTransport
 from trex_core.adapters.persistence.in_memory_intent_store import InMemoryIntentStore
 from trex_core.application.execution import ExecutionService
-from trex_core.application.intents import BuyIntent, IntentService
+from trex_core.application.intents import BuyIntent, IntentService, SellIntent
 from trex_core.application.ledger_outbox import InMemoryLedgerOutbox, LedgerWriter
 from trex_core.application.messaging_handlers import slot_transition_event_payload
 from trex_core.domain.slot import SlotService, SlotState
@@ -27,8 +27,11 @@ def _wire(transport):
     writer = LedgerWriter(outbox, ledger)
     repo.reconcile_universe("default", {"PF-1": "SYM"})
 
-    def emitter(event_type: str, slot) -> None:
+    def emitter(event_type: str, slot, routing_strategy_id: str | None = None) -> None:
-        rk = topology.event_routing_key(event_type, slot.strategy_id or "_", slot.symbol or "_")
+        # Use routing_strategy_id hint when supplied (e.g. sell fills clear
+        # slot.strategy_id via INACTIVE transition — NO_ROUTE fix).
+        effective_sid = routing_strategy_id or getattr(slot, "strategy_id", None) or "_"
+        rk = topology.event_routing_key(event_type, effective_sid, slot.symbol or "_")
         payload = slot_transition_event_payload(slot, event_type=event_type)
         try:
             transport.publish_event(rk, payload)
@@ -124,3 +127,94 @@ def test_publish_failure_does_not_break_execution_loop():
     assert r.ledger_enqueued == 1
     pending = [s for s in repo.list_slots("default") if s.status is SlotState.HOLDING]
     assert len(pending) == 1
+
+
+def test_sell_fill_emits_slot_filled_sell_with_correct_strategy_routing():
+    """Regression: slot_filled_sell must route to strategy queue even after
+    PENDING_SELL->INACTIVE clears strategy_id from the slot row.
+
+    M2 REL-009 feedback: slot_filled_sell was NO_ROUTEd (DLQ) because
+    _on_filled used the post-transition slot (strategy_id=None) for the
+    routing key, producing `slot_filled_sell._.{symbol}` which does not match
+    the `*.{strategy_id}.#` binding pattern of trex.events.strategy.{sid}.
+    """
+    transport = InMemoryTransport()
+    received_sell: list[tuple[str, dict]] = []
+    stop = threading.Event()
+    th = threading.Thread(
+        target=transport.consume_events_forever,
+        args=("trex.events.strategy.A", topology.strategy_event_binding_pattern("A"),
+              lambda rk, p: received_sell.append((rk, p)), stop),
+        daemon=True,
+    )
+    th.start()
+    time.sleep(0.05)
+
+    repo, svc, intents, execution, _ = _wire(transport)
+    # Buy with strategy_id="A"
+    d = intents.submit_buy(BuyIntent(
+        intent_id="b1", qty=10, strategy_id="A", portfolio_id="PF-1",
+    ))
+    execution.run_once()  # buy fill -> holding
+    slot = repo.fetch_slot(d.slot_id)
+    assert slot.status is SlotState.HOLDING
+    assert slot.strategy_id == "A"
+
+    # Sell — strategy_id_for_emit must be captured before INACTIVE clears it.
+    intents.submit_sell(SellIntent(intent_id="s1", slot_id=d.slot_id))
+    execution.run_once()  # sell fill -> inactive
+
+    time.sleep(0.05)
+    stop.set()
+    th.join(timeout=1.0)
+
+    # The slot_filled_sell event must reach the strategy A queue.
+    sell_events = [(rk, p) for rk, p in received_sell if "sell" in rk]
+    assert len(sell_events) == 1, (
+        f"Expected 1 slot_filled_sell event on strategy-A queue, got {len(sell_events)}. "
+        f"All received: {[rk for rk, _ in received_sell]}"
+    )
+    rk, payload = sell_events[0]
+    # Routing key must use strategy_id="A", not "_" (NO_ROUTE sentinel).
+    assert rk == "slot_filled_sell.A.005930", (
+        f"NO_ROUTE regression: routing key is '{rk}', expected 'slot_filled_sell.A.005930'"
+    )
+    assert payload["event_type"] == "slot_filled_sell"
+
+
+def test_sell_fill_dlq_sentinel_absent_when_strategy_id_set():
+    """Routing key must NOT contain '_' as strategy_id token when a real
+    strategy_id was set on the slot (no DLQ sentinel in the key)."""
+    transport = InMemoryTransport()
+    all_published: list[str] = []
+
+    def emitter(event_type: str, slot, routing_strategy_id: str | None = None) -> None:
+        effective_sid = routing_strategy_id or getattr(slot, "strategy_id", None) or "_"
+        rk = topology.event_routing_key(event_type, effective_sid, slot.symbol or "_")
+        all_published.append(rk)
+        payload = slot_transition_event_payload(slot, event_type=event_type)
+        try:
+            transport.publish_event(rk, payload)
+        except Exception:  # noqa: BLE001
+            pass
+
+    repo = InMemorySlotRepository()
+    svc = SlotService(repo)
+    intents = IntentService(svc, repo, InMemoryIntentStore(),
+                            resolve_symbol=lambda pid: "005930")
+    from trex_core.application.ledger_outbox import InMemoryLedgerOutbox
+    outbox = InMemoryLedgerOutbox()
+    execution = ExecutionService(svc, repo, PaperBroker(fill_after_polls=0),
+                                 ledger_outbox=outbox, event_emitter=emitter)
+    repo.reconcile_universe("default", {"PF-1": "SYM"})
+
+    d = intents.submit_buy(BuyIntent(intent_id="b1", qty=10, strategy_id="orb-v1", portfolio_id="PF-1"))
+    execution.run_once()
+    intents.submit_sell(SellIntent(intent_id="s1", slot_id=d.slot_id))
+    execution.run_once()
+
+    # Neither buy nor sell events should use "_" as strategy token.
+    bad_keys = [rk for rk in all_published if "._.".encode() in rk.encode() or rk.split(".")[1] == "_"]
+    assert bad_keys == [], (
+        f"DLQ-sentinel routing keys found (strategy_id=None leak): {bad_keys}"
+    )

trex_core/application/execution.py

@@ -41,7 +41,14 @@ _log = logging.getLogger(__name__)
 # slot transition events through MessagingTransport. Application layer stays
 # transport-agnostic (NFR-1). Slot type is duck-typed to keep the application
 # layer free of domain re-imports (the slot is just used for attribute access).
-SlotEventEmitter = Callable[[str, object], None]  # (event_type, slot) -> None
+#
+# The optional third argument `routing_strategy_id` is a hint for the routing
+# key when the slot's own strategy_id has been cleared by the transition (e.g.
+# PENDING_SELL -> INACTIVE clears strategy_id on the slot row — REQ-057
+# _apply_transition_fields). The emitter MUST prefer this hint over
+# `slot.strategy_id` so that slot_filled_sell events reach the correct
+# per-strategy queue instead of being NO_ROUTEd to the DLQ.
+SlotEventEmitter = Callable[[str, object, Optional[str]], None]  # (event_type, slot, routing_strategy_id) -> None
 
 _MAX_DISPATCH_PER_TICK = 64
 
@@ -288,6 +295,13 @@ class ExecutionService:
 
     def _on_filled(self, slot, fs, side: str, r: ExecutionResult) -> None:
         broker_order_id = slot.broker_order_id  # captured BEFORE clear_dispatch
+        # Capture strategy_id BEFORE the transition clears it.
+        # PENDING_SELL -> INACTIVE clears strategy_id on the slot row
+        # (_apply_transition_fields REQ-057), so we must capture it here to
+        # supply as a routing hint for the slot_filled_sell event — otherwise the
+        # routing key uses "_" and the message is NO_ROUTEd to the DLQ instead
+        # of reaching trex.events.strategy.{strategy_id} (M2 REL-009 feedback).
+        routing_strategy_id = slot.strategy_id
         # REQ-095: resolve fill price with limit_price backfill guard.
         fill_price, _was_backfilled = self._resolve_fill_price(fs, slot)
         if fill_price is None:
@@ -342,7 +356,9 @@ class ExecutionService:
         self._record_ledger(slot, fs, side, broker_order_id, r, resolved_price=fill_price)
         # REQ-021/023: emit event AFTER transition + ledger enqueue (single
         # record subject). Best-effort: a publish failure must not roll back.
-        self._emit_event(event_type, updated)
+        # Pass routing_strategy_id so sell fills route correctly even after
+        # strategy_id is cleared from the slot row (slot_filled_sell NO_ROUTE fix).
+        self._emit_event(event_type, updated, routing_strategy_id)
         self._repo.clear_dispatch(slot.slot_id)  # done with this order
 
     def _on_partial_cancel(self, slot, fs, side: str, r: ExecutionResult) -> None:
@@ -355,6 +371,9 @@ class ExecutionService:
         portion that actually filled.
         """
         broker_order_id = slot.broker_order_id  # captured BEFORE clear_dispatch
+        # Capture strategy_id BEFORE the transition (same reason as _on_filled:
+        # PENDING_SELL -> INACTIVE clears strategy_id, causing NO_ROUTE).
+        routing_strategy_id = slot.strategy_id
         # REQ-095: resolve fill price with limit_price backfill guard.
         fill_price, _was_backfilled = self._resolve_fill_price(fs, slot)
         if fill_price is None:
@@ -395,7 +414,7 @@ class ExecutionService:
         except SlotStateMismatchError:
             return  # slot moved concurrently; next tick reconciles
         self._record_ledger(slot, fs, side, broker_order_id, r, resolved_price=fill_price)
-        self._emit_event(event_type, updated)
+        self._emit_event(event_type, updated, routing_strategy_id)
         self._repo.clear_dispatch(slot.slot_id)
 
     def _check_stale_pending(self, slot, now: datetime, r: ExecutionResult) -> None:
@@ -454,11 +473,11 @@ class ExecutionService:
                 slot.slot_id, slot.broker_order_id,
             )
 
-    def _emit_event(self, event_type: str, slot) -> None:
+    def _emit_event(self, event_type: str, slot, routing_strategy_id: Optional[str] = None) -> None:
         if self._event_emitter is None:
             return
         try:
-            self._event_emitter(event_type, slot)
+            self._event_emitter(event_type, slot, routing_strategy_id)
         except Exception:  # noqa: BLE001 - event publish failure is non-critical
             return
 

trex_core/runtime/daemon.py

@@ -264,10 +264,18 @@ class Daemon:
 
     # ------------------------------------------------------------ event hook
     def _make_event_emitter(self):
-        def emit(event_type: str, slot) -> None:
+        def emit(event_type: str, slot, routing_strategy_id: str | None = None) -> None:
             payload = slot_transition_event_payload(slot, event_type=event_type)
+            # routing_strategy_id is supplied when the transition clears
+            # slot.strategy_id (e.g. PENDING_SELL -> INACTIVE for slot_filled_sell).
+            # Without this hint the routing key would use "_" and the message would
+            # be NO_ROUTEd to the DLQ instead of reaching the strategy queue
+            # (M2 REL-009 feedback / slot_filled NO_ROUTE fix).
+            effective_strategy_id = (
+                routing_strategy_id or getattr(slot, "strategy_id", None) or "_"
+            )
             rk = topology.event_routing_key(
-                event_type, slot.strategy_id or "_", slot.symbol or "_",
+                event_type, effective_strategy_id, slot.symbol or "_",
             )
             try:
                 self.event_publisher.publish_event(rk, payload)