diff --git a/features/event_bus.feature b/features/event_bus.feature index 2875b4f58..1fd540c7e 100644 --- a/features/event_bus.feature +++ b/features/event_bus.feature @@ -163,3 +163,26 @@ Feature: EventBus protocol and domain event emission Scenario: Container wires event_bus into DecisionService When I resolve decision_service from the DI container Then the decision_service should have an event_bus attribute + + # --------------------------------------------------------------------------- + # ReactiveEventBus.close() and context manager (issue #10378) + # --------------------------------------------------------------------------- + + Scenario: ReactiveEventBus.close() completes the RxPY stream + Given a ReactiveEventBus + When bus.close() is called on the bus + Then the bus should be marked as closed + + Scenario: ReactiveEventBus.close() prevents further emit() calls + Given a ReactiveEventBus + When bus.close() is called on the bus + Then emitting after close should raise RuntimeError + + Scenario: ReactiveEventBus.close() is idempotent + Given a ReactiveEventBus + When bus.close() is called twice on the bus + Then no exception should be raised on double close + + Scenario: ReactiveEventBus supports context manager protocol + Given a ReactiveEventBus used as a context manager + Then the bus should be closed after the context exits diff --git a/features/steps/event_bus_steps.py b/features/steps/event_bus_steps.py index 35c1abd93..a414e2ed2 100644 --- a/features/steps/event_bus_steps.py +++ b/features/steps/event_bus_steps.py @@ -477,3 +477,56 @@ def step_decision_service_has_event_bus(ctx: Context) -> None: "DecisionService resolved from DI has no event_bus attribute" ) assert svc.event_bus is not None, "DecisionService.event_bus should not be None" + + +# --------------------------------------------------------------------------- +# ReactiveEventBus.close() and context manager steps (issue #10378) +# --------------------------------------------------------------------------- + + +@when("bus.close() is called on the bus") +def step_close_bus(ctx: Context) -> None: + ctx.bus.close() + + +@then("the bus should be marked as closed") +def step_bus_is_closed(ctx: Context) -> None: + assert ctx.bus._closed, "Expected bus._closed to be True after close()" + + +@then("emitting after close should raise RuntimeError") +def step_emit_after_close_raises(ctx: Context) -> None: + from cleveragents.infrastructure.events.models import DomainEvent + from cleveragents.infrastructure.events.types import EventType + + raised = False + try: + ctx.bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + except RuntimeError: + raised = True + assert raised, "Expected RuntimeError when emitting after close()" + + +@when("bus.close() is called twice on the bus") +def step_close_bus_twice(ctx: Context) -> None: + ctx.bus.close() + ctx.bus.close() # should not raise + + +@then("no exception should be raised on double close") +def step_no_exception_on_double_close(ctx: Context) -> None: + # If we reached here without exception, the test passes + pass + + +@given("a ReactiveEventBus used as a context manager") +def step_given_bus_context_manager(ctx: Context) -> None: + ctx.bus = ReactiveEventBus() + ctx.cm_bus: ReactiveEventBus | None = None + with ctx.bus as bus: + ctx.cm_bus = bus + + +@then("the bus should be closed after the context exits") +def step_bus_closed_after_context(ctx: Context) -> None: + assert ctx.bus._closed, "Expected bus._closed to be True after context manager exit" diff --git a/features/steps/tdd_reactive_event_bus_close_steps.py b/features/steps/tdd_reactive_event_bus_close_steps.py new file mode 100644 index 000000000..8c9e11ba5 --- /dev/null +++ b/features/steps/tdd_reactive_event_bus_close_steps.py @@ -0,0 +1,135 @@ +"""Step definitions for tdd_reactive_event_bus_close.feature. + +TDD tests for issue #10377 / bug #10378: +ReactiveEventBus._subject (RxPY Subject) is never completed — missing +close() method causes RxPY subscriber resource leaks. +""" + +from __future__ import annotations + +from behave import given, then, when # type: ignore[import-untyped] +from behave.runner import Context # type: ignore[import-untyped] + +from cleveragents.infrastructure.events import EventType, ReactiveEventBus +from cleveragents.infrastructure.events.models import DomainEvent + + +def _make_event(event_type_str: str) -> DomainEvent: + return DomainEvent(event_type=EventType(event_type_str)) + + +# --------------------------------------------------------------------------- +# Given +# --------------------------------------------------------------------------- + + +@given("a ReactiveEventBus instance for close testing") +def step_given_bus_for_close(ctx: Context) -> None: + ctx.bus = ReactiveEventBus() + ctx.completed: list[bool] = [] + ctx.close_exception: Exception | None = None + ctx.emit_exception: Exception | None = None + + +@given("a subscriber collecting completion signal from bus stream") +def step_given_completion_subscriber(ctx: Context) -> None: + ctx.completed = [] + ctx.bus.stream.subscribe( + on_next=lambda _: None, + on_error=lambda _: None, + on_completed=lambda: ctx.completed.append(True), + ) + + +@given("a ReactiveEventBus instance used as a context manager") +def step_given_bus_as_context_manager(ctx: Context) -> None: + ctx.bus = ReactiveEventBus() + ctx.context_manager_closed: bool = False + ctx.completed = [] + ctx.bus.stream.subscribe( + on_next=lambda _: None, + on_error=lambda _: None, + on_completed=lambda: ctx.completed.append(True), + ) + + +# --------------------------------------------------------------------------- +# When +# --------------------------------------------------------------------------- + + +@when("bus.close() is called") +def step_when_close_called(ctx: Context) -> None: + try: + ctx.bus.close() + except Exception as exc: + ctx.close_exception = exc + + +@when("an event is emitted after close") +def step_when_emit_after_close(ctx: Context) -> None: + try: + ctx.bus.emit(_make_event("plan.created")) + except RuntimeError as exc: + ctx.emit_exception = exc + + +@when("the context manager exits") +def step_when_context_manager_exits(ctx: Context) -> None: + with ctx.bus: + pass + ctx.context_manager_closed = True + + +@when("bus.close() is called twice") +def step_when_close_called_twice(ctx: Context) -> None: + try: + ctx.bus.close() + ctx.bus.close() + except Exception as exc: + ctx.close_exception = exc + + +# --------------------------------------------------------------------------- +# Then +# --------------------------------------------------------------------------- + + +@then("the subscriber should have received the on_completed signal") +def step_then_completed_signal_received(ctx: Context) -> None: + assert ctx.completed, ( + "Expected on_completed to be called on the RxPY stream after close(), " + "but no completion signal was received" + ) + + +@then("a RuntimeError should be raised") +def step_then_runtime_error_raised(ctx: Context) -> None: + assert isinstance(ctx.emit_exception, RuntimeError), ( + f"Expected RuntimeError when emitting after close(), " + f"got {type(ctx.emit_exception).__name__!r}: {ctx.emit_exception}" + ) + + +@then("bus.close() should have been called automatically") +def step_then_close_called_automatically(ctx: Context) -> None: + assert ctx.context_manager_closed, "Context manager did not exit as expected" + assert ctx.bus._closed, ( + "Expected bus._closed to be True after context manager exit, " + "indicating close() was called" + ) + + +@then("the RxPY stream should be completed") +def step_then_rxpy_stream_completed(ctx: Context) -> None: + assert ctx.completed, ( + "Expected on_completed to be called on the RxPY stream after context " + "manager exit, but no completion signal was received" + ) + + +@then("no exception should be raised on the second close call") +def step_then_no_exception_on_second_close(ctx: Context) -> None: + assert ctx.close_exception is None, ( + f"Expected no exception on second close(), got: {ctx.close_exception}" + ) diff --git a/features/tdd_reactive_event_bus_close.feature b/features/tdd_reactive_event_bus_close.feature new file mode 100644 index 000000000..59a2afdfd --- /dev/null +++ b/features/tdd_reactive_event_bus_close.feature @@ -0,0 +1,39 @@ +@tdd_issue @tdd_issue_10377 +Feature: TDD Issue #10377 — ReactiveEventBus missing close() method causes RxPY subscriber resource leaks + As a developer using ReactiveEventBus with RxPY operators + I want the bus to provide a close() method that completes the RxPY Subject + So that buffering and aggregation operators can finalize their state without leaking resources + + This test suite captures bug #10378 / TDD issue #10377. + The tests verify that ReactiveEventBus.close() completes the RxPY stream, + that emit() raises RuntimeError after close(), and that the bus supports + the context manager protocol for automatic cleanup. + + See CONTRIBUTING.md > Bug Fix Workflow > TDD Issue Test Tags. + + @tdd_issue @tdd_issue_10377 + Scenario: ReactiveEventBus.close() completes the RxPY stream + Given a ReactiveEventBus instance for close testing + And a subscriber collecting completion signal from bus stream + When bus.close() is called + Then the subscriber should have received the on_completed signal + + @tdd_issue @tdd_issue_10377 + Scenario: ReactiveEventBus.close() prevents further emit() calls + Given a ReactiveEventBus instance for close testing + When bus.close() is called + And an event is emitted after close + Then a RuntimeError should be raised + + @tdd_issue @tdd_issue_10377 + Scenario: ReactiveEventBus implements context manager protocol + Given a ReactiveEventBus instance used as a context manager + When the context manager exits + Then bus.close() should have been called automatically + And the RxPY stream should be completed + + @tdd_issue @tdd_issue_10377 + Scenario: ReactiveEventBus.close() is idempotent + Given a ReactiveEventBus instance for close testing + When bus.close() is called twice + Then no exception should be raised on the second close call diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index 15af69bb7..41884990f 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -21,6 +21,7 @@ from __future__ import annotations import contextlib from collections import deque from collections.abc import Callable +from types import TracebackType import structlog from rx import operators as ops @@ -50,6 +51,7 @@ class ReactiveEventBus: _stream: Read-only Observable view over ``_subject``. _subscriptions: Per-event-type handler lists. _audit_log: Volatile in-memory log of all emitted events. + _closed: Whether :meth:`close` has been called. """ def __init__(self, max_audit_log_size: int | None = None) -> None: @@ -60,6 +62,7 @@ class ReactiveEventBus: self._stream: Observable = self._subject.pipe(ops.map(_identity)) self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {} self._audit_log: deque[DomainEvent] = deque(maxlen=max_audit_log_size) + self._closed: bool = False # ------------------------------------------------------------------ # Public API (satisfies EventBus protocol) @@ -107,12 +110,19 @@ class ReactiveEventBus: Raises: TypeError: If *event* is not a :class:`DomainEvent`. + RuntimeError: If :meth:`close` has already been called. """ if not isinstance(event, DomainEvent): raise TypeError( f"event must be a DomainEvent, got {type(event).__name__!r}" ) + if self._closed: + raise RuntimeError( + "ReactiveEventBus.emit() called after close() — " + "the bus has been shut down and the RxPY Subject is completed" + ) + try: self._subject.on_next(event) except Exception as exc: @@ -135,7 +145,6 @@ class ReactiveEventBus: handler=getattr(handler, "__qualname__", repr(handler)), error_type=type(exc).__name__, error=str(exc), - exc_info=True, ) def subscribe( @@ -175,16 +184,49 @@ class ReactiveEventBus: return self._stream def close(self) -> None: - """Signal completion on the reactive stream and clear all subscriptions. + """Complete the RxPY stream and prevent further event emission. - Call this when the bus is no longer needed (e.g. in test teardown) - to release any RxPY Subject resources and prevent subscription leaks - between test scenarios. + Signals ``on_completed()`` to all RxPY subscribers, allowing them to + finalize their state (e.g. buffering operators flush, aggregation + operators emit their final result). After ``close()``, any call to + :meth:`emit` raises :exc:`RuntimeError`. + + This method is idempotent — calling it more than once is safe. + + Call this when the bus is no longer needed (e.g. in test teardown or + application shutdown) to release RxPY Subject resources and prevent + subscription leaks between test scenarios. """ + if self._closed: + return + self._closed = True with contextlib.suppress(Exception): self._subject.on_completed() self._subscriptions.clear() self._audit_log.clear() + def __enter__(self) -> ReactiveEventBus: + """Support use as a context manager. + + Returns: + The bus instance itself. + """ + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + """Call :meth:`close` on context manager exit. + + Args: + exc_type: Exception type, if any. + exc_val: Exception value, if any. + exc_tb: Exception traceback, if any. + """ + self.close() + __all__ = ["ReactiveEventBus"]