feat(messaging): OPS-005 이벤트 발행 연결 신뢰성 (BAT-047)

발행 전용 연결의 idle heartbeat 사망 후 첫 발행 유실 결함 제거. - REQ-091 발행 재시도 + 강제 재연결(_force_reconnect, is_open 불신) - REQ-092 idle keepalive(execution-loop 단일스레드 틱, process_data_events time_limit=0) - REQ-093 DLQ 가시성 + emit 실패 ERROR 경보(구조화 로그, dcy-029 재사용) - REQ-094 baseline 문서 정합(runtime-flows/operations/project-structure) 독립 opus QC PASS-WITH-NOTES → 시정 2건(ruff F401, AC-3 프로덕션 경로) 완료. 554 passed/47 skipped/회귀 0/ruff clean. NFR-1 wire 무회귀(interface-contract diff 0). BAT-047 release-candidate, REQ-091~094 Implemented. dcy-031. M2 트랙B. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VHtM5Ws1HKCKQ4JGi6GRb5
2026-06-19 22:39:43 +09:00
parent bc79f60274
commit d182464b7a
19 changed files with 1067 additions and 43 deletions
@@ -0,0 +1,455 @@
+"""BAT-047 — Event publisher connection reliability tests (REQ-091/092/093).
+
+All tests run without a live RabbitMQ broker using unittest.mock.  The
+RabbitMQTransport is exercised through its public interface; pika internals
+are replaced by lightweight mocks so the retry / keepalive / DLQ logic can
+be unit-tested deterministically.
+
+AC coverage:
+  REQ-091 AC-1 : half-open → reconnect + retry → success (no error raised)
+  REQ-091 AC-2 : half-open + retry also fails → TransportUnavailableError
+  REQ-091 AC-3 : timeout budget — _force_reconnect uses connect_timeout=2.0
+  REQ-091 AC-4 : normal publish (conn alive) — no retry overhead
+  REQ-091 AC-5 : consume/RPC paths untouched by this change (interface check)
+  REQ-091 AC-6 : broker restart scenario — reconnect + retry succeeds
+  REQ-092 AC-1 : keepalive_tick calls process_data_events on live conn
+  REQ-092 AC-2 : keepalive_tick executes only on owning object (no lock escape)
+  REQ-092 AC-3 : simultaneous publish + keepalive — no crash (single-thread)
+  REQ-092 AC-4 : keepalive non-blocking (time_limit=0 passed to pika)
+  REQ-092 AC-5 : consume/RPC paths unaffected (keepalive not called there)
+  REQ-093 AC-1 : unroutable → _publish_dlq called + ERROR log emitted
+  REQ-093 AC-3 : emit ERROR on final publish failure (daemon layer)
+  REQ-093 AC-4 : _publish_dlq itself fails → ERROR log, no silent loss
+  REQ-093 AC-6 : consume/RPC paths unaffected (DLQ change local to publish)
+"""
+from __future__ import annotations
+
+import logging
+from unittest.mock import MagicMock, patch
+
+import pika.exceptions
+import pytest
+
+from trex_core.adapters.messaging.rabbitmq import RabbitMQTransport
+from trex_core.application.messaging import TransportUnavailableError
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_transport() -> RabbitMQTransport:
+    """Return a RabbitMQTransport instance without connecting to a real broker."""
+    return RabbitMQTransport("amqp://localhost/", timeout=5.0)
+
+
+def _inject_connected(t: RabbitMQTransport, conn=None, channel=None):
+    """Put the transport into CONNECTED state with mock pika objects."""
+    from trex_core.application.messaging import TransportHealth
+    if conn is None:
+        conn = MagicMock()
+        conn.is_open = True
+    if channel is None:
+        channel = MagicMock()
+        channel.is_open = True
+    t._conn = conn
+    t._channel = channel
+    t._health = TransportHealth.CONNECTED
+    return conn, channel
+
+
+# ---------------------------------------------------------------------------
+# REQ-091 AC-1: half-open → reconnect + retry → success
+# ---------------------------------------------------------------------------
+
+class TestPublishRetryOnHalfOpen:
+    """REQ-091 AC-1: publish succeeds after forced reconnect."""
+
+    def test_half_open_first_attempt_fails_retry_succeeds(self, caplog):
+        """Simulates a half-open connection: is_open=True but basic_publish raises
+        AMQPConnectionError on first call.  The transport must reconnect and retry,
+        and publish_event must NOT raise.
+        """
+        t = _make_transport()
+        mock_conn, mock_ch = _inject_connected(t)
+
+        call_count = {"n": 0}
+
+        def fake_basic_publish(**kwargs):
+            call_count["n"] += 1
+            if call_count["n"] == 1:
+                raise pika.exceptions.AMQPConnectionError("connection reset")
+            # Second call (after reconnect) succeeds silently.
+
+        mock_ch.basic_publish.side_effect = fake_basic_publish
+
+        reconnect_called = {"yes": False}
+
+        def fake_force_reconnect():
+            reconnect_called["yes"] = True
+            # After reconnect, the channel is fresh and functional.
+            new_ch = MagicMock()
+            new_ch.is_open = True
+            new_ch.basic_publish.return_value = None  # success
+            t._channel = new_ch
+            from trex_core.application.messaging import TransportHealth
+            t._health = TransportHealth.CONNECTED
+
+        with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect):
+            # Must not raise.
+            t.publish_event("slot_filled_buy.A.005930", {"slot": "s1"})
+
+        assert reconnect_called["yes"], "_force_reconnect must be called on first failure"
+
+
+# ---------------------------------------------------------------------------
+# REQ-091 AC-2: retry also fails → TransportUnavailableError raised
+# ---------------------------------------------------------------------------
+
+class TestPublishRetryExhausted:
+    """REQ-091 AC-2: when reconnect+retry also fails, error is propagated."""
+
+    def test_retry_exhausted_raises_transport_unavailable(self):
+        t = _make_transport()
+        mock_conn, mock_ch = _inject_connected(t)
+
+        mock_ch.basic_publish.side_effect = pika.exceptions.AMQPConnectionError("gone")
+
+        def fake_force_reconnect():
+            from trex_core.application.messaging import TransportHealth
+            new_ch = MagicMock()
+            new_ch.is_open = True
+            new_ch.basic_publish.side_effect = pika.exceptions.AMQPConnectionError(
+                "still gone"
+            )
+            t._channel = new_ch
+            t._health = TransportHealth.CONNECTED
+
+        with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect):
+            with pytest.raises(TransportUnavailableError) as exc_info:
+                t.publish_event("slot_filled_buy.A.005930", {"slot": "s1"})
+
+        assert "reconnect+retry" in str(exc_info.value)
+
+
+# ---------------------------------------------------------------------------
+# REQ-091 AC-3: timeout budget — _force_reconnect uses connect_timeout=2.0
+# ---------------------------------------------------------------------------
+
+class TestForceReconnectTimeout:
+    """REQ-091 AC-3 / NFR-2: _force_reconnect passes connect_timeout=2.0."""
+
+    def test_force_reconnect_uses_2s_timeout(self):
+        t = _make_transport()
+        _inject_connected(t)
+
+        connect_calls: list = []
+
+        def fake_connect(connect_timeout=None):
+            connect_calls.append(connect_timeout)
+            from trex_core.application.messaging import TransportHealth
+            new_conn = MagicMock()
+            new_conn.is_open = True
+            new_ch = MagicMock()
+            new_ch.is_open = True
+            t._conn = new_conn
+            t._channel = new_ch
+            t._health = TransportHealth.CONNECTED
+
+        with patch.object(t, "_connect", side_effect=fake_connect):
+            t._force_reconnect()
+
+        assert len(connect_calls) == 1
+        assert connect_calls[0] == 2.0, (
+            f"_force_reconnect must pass connect_timeout=2.0, got {connect_calls[0]}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# REQ-091 AC-4: normal publish (conn alive) — no retry path triggered
+# ---------------------------------------------------------------------------
+
+class TestNormalPublishNoRetry:
+    """REQ-091 AC-4: successful first-attempt publish does not invoke retry."""
+
+    def test_normal_publish_does_not_call_force_reconnect(self):
+        t = _make_transport()
+        _inject_connected(t)
+        t._channel.basic_publish.return_value = None  # success
+
+        with patch.object(t, "_force_reconnect") as mock_fr:
+            t.publish_event("slot_filled_buy.A.005930", {"ok": True})
+
+        mock_fr.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# REQ-091 AC-6: broker restart — force reconnect + retry succeeds
+# ---------------------------------------------------------------------------
+
+class TestBrokerRestartScenario:
+    """REQ-091 AC-6: after broker restart the transport reconnects and retries."""
+
+    def test_broker_restart_publish_recovers(self):
+        t = _make_transport()
+        _inject_connected(t)
+
+        # Simulate broker restart: first publish raises OSError (TCP reset).
+        t._channel.basic_publish.side_effect = OSError("connection refused")
+
+        recovered = {"yes": False}
+
+        def fake_force_reconnect():
+            from trex_core.application.messaging import TransportHealth
+            new_ch = MagicMock()
+            new_ch.is_open = True
+            new_ch.basic_publish.return_value = None
+            t._channel = new_ch
+            t._health = TransportHealth.CONNECTED
+            recovered["yes"] = True
+
+        with patch.object(t, "_force_reconnect", side_effect=fake_force_reconnect):
+            # publish_event must succeed without raising.
+            t.publish_event("slot_filled_buy.A.005930", {"ev": "restart"})
+
+        assert recovered["yes"]
+
+
+# ---------------------------------------------------------------------------
+# REQ-092 AC-1: keepalive_tick calls process_data_events on live conn
+# ---------------------------------------------------------------------------
+
+class TestKeepaliveTick:
+    """REQ-092 AC-1: keepalive_tick delegates to conn.process_data_events."""
+
+    def test_keepalive_tick_calls_process_data_events_on_live_connection(self):
+        t = _make_transport()
+        mock_conn, _ = _inject_connected(t)
+        mock_conn.is_open = True
+
+        t.keepalive_tick()
+
+        mock_conn.process_data_events.assert_called_once_with(time_limit=0)
+
+    def test_keepalive_tick_noop_when_disconnected(self):
+        """REQ-092 AC-1 variant: no-op if connection is None (shutdown path)."""
+        t = _make_transport()
+        # _conn is None by default — keepalive_tick must not raise.
+        t.keepalive_tick()  # should return silently
+
+    def test_keepalive_tick_noop_when_conn_closed(self):
+        """keepalive_tick skips when is_open=False."""
+        t = _make_transport()
+        mock_conn = MagicMock()
+        mock_conn.is_open = False
+        t._conn = mock_conn
+        from trex_core.application.messaging import TransportHealth
+        t._health = TransportHealth.DISCONNECTED
+
+        t.keepalive_tick()
+
+        mock_conn.process_data_events.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# REQ-092 AC-4: keepalive non-blocking — time_limit=0
+# ---------------------------------------------------------------------------
+
+class TestKeepaliveTimeLimitZero:
+    """REQ-092 AC-4: process_data_events is called with time_limit=0."""
+
+    def test_keepalive_uses_time_limit_zero(self):
+        t = _make_transport()
+        mock_conn, _ = _inject_connected(t)
+        mock_conn.is_open = True
+
+        t.keepalive_tick()
+
+        args, kwargs = mock_conn.process_data_events.call_args
+        assert kwargs.get("time_limit", args[0] if args else None) == 0
+
+
+# ---------------------------------------------------------------------------
+# REQ-092 AC-3: publish + keepalive in same thread — no crash
+# ---------------------------------------------------------------------------
+
+class TestKeepaliveAndPublishSameThread:
+    """REQ-092 AC-3: calling keepalive then publish in the same thread is safe."""
+
+    def test_keepalive_then_publish_sequential_no_error(self):
+        t = _make_transport()
+        mock_conn, _ = _inject_connected(t)
+        mock_conn.is_open = True
+        t._channel.basic_publish.return_value = None
+
+        # Sequential in same thread — simulates execution-loop tick.
+        t.keepalive_tick()
+        t.publish_event("slot_filled_buy.A.005930", {"seq": 1})
+
+        mock_conn.process_data_events.assert_called_once_with(time_limit=0)
+        t._channel.basic_publish.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# REQ-092 AC-5: consume/RPC paths are unaffected by keepalive
+# ---------------------------------------------------------------------------
+
+class TestKeepaliveDoesNotAffectConsumePaths:
+    """REQ-092 AC-5: keepalive is specific to publish path; consume API unchanged."""
+
+    def test_keepalive_tick_not_present_on_consume_rpc_forever_signature(self):
+        """The consume_rpc_forever signature is unchanged."""
+        import inspect
+        sig = inspect.signature(RabbitMQTransport.consume_rpc_forever)
+        params = list(sig.parameters)
+        assert "queue" in params
+        assert "handler" in params
+        assert "stop" in params
+
+
+# ---------------------------------------------------------------------------
+# REQ-093 AC-1: unroutable → _publish_dlq called + ERROR log emitted
+# ---------------------------------------------------------------------------
+
+class TestDlqVisibilityOnUnroutable:
+    """REQ-093 AC-1: unroutable event triggers _publish_dlq + ERROR log."""
+
+    def test_unroutable_calls_publish_dlq_and_logs_error(self, caplog):
+        t = _make_transport()
+        _inject_connected(t)
+
+        t._channel.basic_publish.side_effect = pika.exceptions.UnroutableError([])
+
+        dlq_calls: list = []
+
+        def fake_dlq(rk, payload, *, reason):
+            dlq_calls.append((rk, reason))
+
+        with patch.object(t, "_publish_dlq", side_effect=fake_dlq):
+            with caplog.at_level(logging.DEBUG):
+                # Must not raise — unroutable is handled via DLQ.
+                t.publish_event("slot_filled_buy.A.005930", {"x": 1})
+
+        assert len(dlq_calls) == 1
+        assert dlq_calls[0][1] == "unroutable"
+
+    def test_publish_dlq_logs_error_on_entry(self, caplog):
+        """REQ-093 AC-1: _publish_dlq itself logs ERROR when DLQ entry is created."""
+        t = _make_transport()
+        _inject_connected(t)
+        # Make DLQ publish succeed (no-op).
+        t._channel.basic_publish.return_value = None
+
+        with caplog.at_level(logging.ERROR, logger="trex_core.adapters.messaging.rabbitmq"):
+            t._publish_dlq("slot_filled_buy.A.005930", {"x": 1}, reason="unroutable")
+
+        errors = [r for r in caplog.records if r.levelno >= logging.ERROR]
+        assert any("DLQ" in r.message for r in errors), (
+            "Expected ERROR log with 'DLQ' when _publish_dlq is called"
+        )
+
+
+# ---------------------------------------------------------------------------
+# REQ-093 AC-4: _publish_dlq itself fails → ERROR log, no silent loss
+# ---------------------------------------------------------------------------
+
+class TestDlqPublishFailureLogsError:
+    """REQ-093 AC-4: when DLQ publish itself raises, ERROR is logged."""
+
+    def test_dlq_publish_failure_logs_error_not_warning(self, caplog):
+        t = _make_transport()
+        _inject_connected(t)
+
+        # First basic_publish for DLQ raises (after _ensure_connected succeeds).
+        t._channel.basic_publish.side_effect = Exception("DLQ broker down")
+
+        with caplog.at_level(logging.ERROR, logger="trex_core.adapters.messaging.rabbitmq"):
+            # _publish_dlq must swallow the exception but log ERROR.
+            t._publish_dlq("slot_filled_buy.A.005930", {"x": 1}, reason="unroutable")
+
+        error_records = [r for r in caplog.records if r.levelno >= logging.ERROR]
+        assert len(error_records) >= 1
+        # The DLQ publish failure must be ERROR, not WARNING.
+        dlq_fail_records = [r for r in error_records if "DLQ" in r.message]
+        assert dlq_fail_records, "Expected ERROR log mentioning 'DLQ' on publish failure"
+
+
+# ---------------------------------------------------------------------------
+# REQ-093 AC-3: emit ERROR on final publish failure (daemon layer)
+# ---------------------------------------------------------------------------
+
+class TestEmitErrorOnFinalPublishFailure:
+    """REQ-093 AC-3: _make_event_emitter logs ERROR (not WARNING) on failure."""
+
+    def test_emit_logs_error_on_transport_unavailable(self, caplog):
+        """The emitter created by _make_event_emitter must log ERROR when
+        publish_event raises TransportUnavailableError after retry.
+
+        Uses the production Daemon._make_event_emitter() directly (via a
+        lightweight stub instance) so that any change to the production error
+        path is caught as a regression.
+        """
+        from trex_core.runtime.daemon import Daemon
+        from trex_core.application.messaging import TransportUnavailableError
+
+        # Construct a minimal Daemon stub without triggering __init__ (which
+        # requires DB / config / broker connections).  Only event_publisher is
+        # accessed by _make_event_emitter, so that is the only attribute we set.
+        daemon_stub = object.__new__(Daemon)
+        fake_publisher = MagicMock()
+        fake_publisher.publish_event.side_effect = TransportUnavailableError("no broker")
+        daemon_stub.event_publisher = fake_publisher
+
+        # Obtain the emitter from the production code path.
+        emit = daemon_stub._make_event_emitter()
+
+        # Minimal slot stub.
+        slot = MagicMock()
+        slot.strategy_id = "A"
+        slot.symbol = "005930"
+        slot.slot_id = "slot-1"
+        slot.status = MagicMock()
+        slot.status.value = "holding"
+        slot.updated_at = None
+        slot.portfolio_id = "PF-1"
+        slot.entry_price = None
+        slot.qty = 10
+        slot.instance_key = "default"
+        slot.actor = None
+        slot.reason = None
+
+        with caplog.at_level(logging.ERROR, logger="trex_core.runtime.daemon"):
+            emit("slot_filled_buy", slot)
+
+        error_records = [
+            r for r in caplog.records
+            if r.levelno >= logging.ERROR and "event publish failed" in r.message
+        ]
+        assert error_records, (
+            "Expected ERROR log 'event publish failed' from Daemon._make_event_emitter"
+        )
+        # Must NOT be mere WARNING.
+        warn_only = [
+            r for r in caplog.records
+            if r.levelno == logging.WARNING and "event publish failed" in r.message
+        ]
+        assert not warn_only, "publish failure must not be WARNING; must be ERROR"
+
+
+# ---------------------------------------------------------------------------
+# REQ-093 AC-6: consume/RPC paths unaffected
+# ---------------------------------------------------------------------------
+
+class TestConsumePaths:
+    """REQ-093 AC-6 / REQ-091 AC-5: consume/RPC interface signature unchanged."""
+
+    def test_rabbitmq_transport_has_consume_methods(self):
+        t = _make_transport()
+        assert callable(getattr(t, "consume_rpc_forever", None))
+        assert callable(getattr(t, "consume_events_forever", None))
+        assert callable(getattr(t, "call_rpc", None))
+        assert callable(getattr(t, "declare_topology", None))
+
+    def test_keepalive_tick_exists_as_callable(self):
+        t = _make_transport()
+        assert callable(getattr(t, "keepalive_tick", None))