diff --git a/.forgejo/workflows/master.yml b/.forgejo/workflows/master.yml index bb14f9ee0..862103661 100644 --- a/.forgejo/workflows/master.yml +++ b/.forgejo/workflows/master.yml @@ -3,8 +3,6 @@ name: CI on: push: branches: [master, develop] - pull_request: - branches: [master, develop] vars: docker_prefix: "http://harbor.cleverthis.com/docker/" diff --git a/CHANGELOG.md b/CHANGELOG.md index 796db5544..199cc79a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,18 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Fixed + +- **EventBus unsubscribe() — memory leak fix** (#10356): Added `unsubscribe()` method + to the `EventBus` protocol and both implementations (`ReactiveEventBus`, + `LoggingEventBus`). Previously, once a handler was registered via `subscribe()` it + could never be removed, causing bound-method handler references to keep owning + objects alive indefinitely in long-running processes. The new `unsubscribe()` method + removes the handler from `_subscriptions`; unsubscribing a handler that was never + registered is a no-op. TDD regression scenarios tagged `@tdd_issue @tdd_issue_10354` + verify unsubscription behaviour and garbage-collection of owning objects. + + ### Changed - **Diagnostics spec examples expanded to all 9 providers** (#5320): Updated the diff --git a/features/event_bus.feature b/features/event_bus.feature index 2875b4f58..759201bc9 100644 --- a/features/event_bus.feature +++ b/features/event_bus.feature @@ -163,3 +163,52 @@ 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 + + # --------------------------------------------------------------------------- + # unsubscribe() — TDD issue #10354 / bug fix #10356 + # --------------------------------------------------------------------------- + + @tdd_issue @tdd_issue_10354 + Scenario: ReactiveEventBus supports unsubscribing a handler + Given a ReactiveEventBus + When I subscribe to "plan.created" events + And I unsubscribe the handler from "plan.created" events + And I emit a "plan.created" DomainEvent + Then the handler should have received 0 event + + @tdd_issue @tdd_issue_10354 + Scenario: LoggingEventBus supports unsubscribing a handler + Given a LoggingEventBus + When I subscribe to "plan.created" events + And I unsubscribe the handler from "plan.created" events + And I emit a "plan.created" DomainEvent + Then the handler should have received 0 event + + @tdd_issue @tdd_issue_10354 + Scenario: Unsubscribing a non-registered handler is a no-op for ReactiveEventBus + Given a ReactiveEventBus + Then unsubscribing a handler that was never registered should not raise + + @tdd_issue @tdd_issue_10354 + Scenario: Unsubscribing a non-registered handler is a no-op for LoggingEventBus + Given a LoggingEventBus + Then unsubscribing a handler that was never registered should not raise + + @tdd_issue @tdd_issue_10354 + Scenario: Unsubscribed ReactiveEventBus handler does not prevent garbage collection + Given a ReactiveEventBus + When I subscribe a bound method handler for "plan.created" events + And I unsubscribe the bound method handler from "plan.created" events + And I delete the owning component reference + Then the component should be garbage collected + + @tdd_issue @tdd_issue_10354 + Scenario: Unsubscribed LoggingEventBus handler does not prevent garbage collection + Given a LoggingEventBus + When I subscribe a bound method handler for "plan.created" events + And I unsubscribe the bound method handler from "plan.created" events + And I delete the owning component reference + Then the component should be garbage collected + + Scenario: EventBus Protocol unsubscribe stub is executable + Then the EventBus Protocol unsubscribe stub should be callable diff --git a/features/steps/actor_run_signature_resolve_steps.py b/features/steps/actor_run_signature_resolve_steps.py index ab6bc56c8..0ced2b52c 100644 --- a/features/steps/actor_run_signature_resolve_steps.py +++ b/features/steps/actor_run_signature_resolve_steps.py @@ -15,7 +15,6 @@ from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock, patch -import click import typer import yaml from behave import then, when @@ -165,7 +164,7 @@ def step_resolve_with_no_config_data(context: Any) -> None: try: resolve_config_files("local/empty-actor", []) context.resolve_exit_code = 0 - except (SystemExit, click.exceptions.Exit) as exc: + except (SystemExit, typer.Exit) as exc: context.resolve_exit_code = getattr( exc, "exit_code", getattr(exc, "code", 1) ) @@ -209,7 +208,7 @@ def step_resolve_unknown_actor_directly(context: Any) -> None: try: resolve_config_files("nonexistent/actor", []) context.resolve_exit_code = 0 - except (SystemExit, click.exceptions.Exit) as exc: + except (SystemExit, typer.Exit) as exc: context.resolve_exit_code = getattr( exc, "exit_code", getattr(exc, "code", 1) ) @@ -260,7 +259,7 @@ def step_resolve_with_empty_config_blob(context: Any) -> None: try: resolve_config_files("local/empty-blob-actor", []) context.resolve_exit_code = 0 - except (SystemExit, click.exceptions.Exit) as exc: + except (SystemExit, typer.Exit) as exc: context.resolve_exit_code = getattr( exc, "exit_code", getattr(exc, "code", 1) ) @@ -356,7 +355,7 @@ def step_resolve_with_unserializable_config_blob(context: Any) -> None: try: resolve_config_files("local/bad-blob-actor", []) context.resolve_exit_code = 0 - except (SystemExit, click.exceptions.Exit) as exc: + except (SystemExit, typer.Exit) as exc: context.resolve_exit_code = getattr( exc, "exit_code", getattr(exc, "code", 1) ) diff --git a/features/steps/actor_run_signature_security_steps.py b/features/steps/actor_run_signature_security_steps.py index 7e93caf7a..fedfa92cd 100644 --- a/features/steps/actor_run_signature_security_steps.py +++ b/features/steps/actor_run_signature_security_steps.py @@ -6,7 +6,6 @@ import contextlib from typing import Any from unittest.mock import MagicMock, patch -import click import typer from behave import then, when @@ -49,7 +48,7 @@ def step_resolve_with_control_character_name(context: Any) -> None: try: resolve_config_files(actor_name, []) context.resolve_exit_code = 0 - except (SystemExit, click.exceptions.Exit) as exc: + except (SystemExit, typer.Exit) as exc: context.resolve_exit_code = getattr( exc, "exit_code", getattr(exc, "code", 1) ) diff --git a/features/steps/event_bus_steps.py b/features/steps/event_bus_steps.py index 35c1abd93..67a8011b4 100644 --- a/features/steps/event_bus_steps.py +++ b/features/steps/event_bus_steps.py @@ -477,3 +477,83 @@ 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" + + +# --------------------------------------------------------------------------- +# unsubscribe() steps — TDD issue #10354 / bug fix #10356 +# --------------------------------------------------------------------------- + + +@when('I unsubscribe the handler from "{et}" events') +def step_unsubscribe_handler(ctx: Context, et: str) -> None: + assert ctx.collectors, "No collectors registered — subscribe first" + ctx.bus.unsubscribe(EventType(et), ctx.collectors[0]) + + +@then("unsubscribing a handler that was never registered should not raise") +def step_unsubscribe_never_registered(ctx: Context) -> None: + raised = False + try: + ctx.bus.unsubscribe(EventType.PLAN_CREATED, lambda e: None) + except Exception: + raised = True + assert not raised, ( + "Expected no exception when unsubscribing an unregistered handler" + ) + + +class _GCComponent: + """Minimal component whose bound method can be subscribed as a handler.""" + + def __init__(self) -> None: + self.received: list[DomainEvent] = [] + + def handle(self, event: DomainEvent) -> None: + self.received.append(event) + + +@when('I subscribe a bound method handler for "{et}" events') +def step_subscribe_bound_method(ctx: Context, et: str) -> None: + import weakref + + ctx.gc_component = _GCComponent() + ctx.gc_component_ref: weakref.ref[_GCComponent] = weakref.ref(ctx.gc_component) + ctx.bound_handler = ctx.gc_component.handle + ctx.bus.subscribe(EventType(et), ctx.bound_handler) + + +@when('I unsubscribe the bound method handler from "{et}" events') +def step_unsubscribe_bound_method(ctx: Context, et: str) -> None: + ctx.bus.unsubscribe(EventType(et), ctx.bound_handler) + + +@when("I delete the owning component reference") +def step_delete_component_ref(ctx: Context) -> None: + import gc + + del ctx.gc_component + del ctx.bound_handler + gc.collect() + + +@then("the component should be garbage collected") +def step_component_garbage_collected(ctx: Context) -> None: + assert ctx.gc_component_ref() is None, ( + "Component was NOT garbage collected — the bus still holds a strong reference" + ) + + +@then("the EventBus Protocol unsubscribe stub should be callable") +def step_protocol_unsubscribe_stub_callable(ctx: Context) -> None: + """Exercise Protocol unsubscribe stub directly to ensure 100% coverage.""" + + obj = type( + "_BareImpl", + (), + { + "emit": lambda self, event: None, + "subscribe": lambda self, et, h: None, + "unsubscribe": lambda self, et, h: None, + }, + )() + obj.unsubscribe(EventType.PLAN_CREATED, lambda e: None) diff --git a/features/steps/fix_then_revalidate_coverage_boost_steps.py b/features/steps/fix_then_revalidate_coverage_boost_steps.py index c900cafa1..e3665be0c 100644 --- a/features/steps/fix_then_revalidate_coverage_boost_steps.py +++ b/features/steps/fix_then_revalidate_coverage_boost_steps.py @@ -547,6 +547,13 @@ class _MockEventBus: ) -> None: pass # Not needed for this test + def unsubscribe( + self, + event_type: EventType, + handler: Any, + ) -> None: + pass # Not needed for this test + class _FailingEventBus: """Mock EventBus whose emit always raises.""" @@ -561,6 +568,13 @@ class _FailingEventBus: ) -> None: pass + def unsubscribe( + self, + event_type: EventType, + handler: Any, + ) -> None: + pass + # --------------------------------------------------------------------------- # Given steps for new review-fix scenarios diff --git a/features/steps/fix_then_revalidate_coverage_r3_steps.py b/features/steps/fix_then_revalidate_coverage_r3_steps.py index d6ec44448..b0a9fb61f 100644 --- a/features/steps/fix_then_revalidate_coverage_r3_steps.py +++ b/features/steps/fix_then_revalidate_coverage_r3_steps.py @@ -118,6 +118,13 @@ class _FailingEventBus: ) -> None: pass + def unsubscribe( + self, + event_type: EventType, + handler: Any, + ) -> None: + pass + # --------------------------------------------------------------------------- # Given steps diff --git a/src/cleveragents/cli/commands/actor.py b/src/cleveragents/cli/commands/actor.py index 9bcaac7e6..fe0c6e006 100644 --- a/src/cleveragents/cli/commands/actor.py +++ b/src/cleveragents/cli/commands/actor.py @@ -185,7 +185,7 @@ def run( except UnsafeConfigurationError as exc: typer.echo(f"Error: {exc}", err=True) raise typer.Exit(code=1) from exc - except click.exceptions.Exit: + except (click.exceptions.Exit, typer.Exit): raise except CleverAgentsError as exc: typer.echo(f"Error: {exc}", err=True) diff --git a/src/cleveragents/cli/commands/actor_run.py b/src/cleveragents/cli/commands/actor_run.py index 14b2d2cf2..0131bd692 100644 --- a/src/cleveragents/cli/commands/actor_run.py +++ b/src/cleveragents/cli/commands/actor_run.py @@ -159,7 +159,7 @@ def run( except UnsafeConfigurationError as exc: typer.echo(f"Error: {exc}", err=True) raise typer.Exit(code=1) from exc - except click.exceptions.Exit: + except (click.exceptions.Exit, typer.Exit): raise except CleverAgentsError as exc: typer.echo(f"Error: {exc}", err=True) diff --git a/src/cleveragents/infrastructure/events/logging_bus.py b/src/cleveragents/infrastructure/events/logging_bus.py index 307bf4ed3..b3d76f94b 100644 --- a/src/cleveragents/infrastructure/events/logging_bus.py +++ b/src/cleveragents/infrastructure/events/logging_bus.py @@ -13,6 +13,7 @@ Based on: from __future__ import annotations +import contextlib from collections.abc import Callable import structlog @@ -105,5 +106,30 @@ class LoggingEventBus: raise TypeError("handler must be callable") self._subscriptions.setdefault(event_type, []).append(handler) + def unsubscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> None: + """Remove *handler* from the list of subscribers for *event_type*. + + If *handler* was not registered for *event_type*, this is a no-op — + no exception is raised. This makes it safe to call during component + teardown without tracking whether :meth:`subscribe` was ever called. + + Removing a handler releases the strong reference held by + ``_subscriptions``, allowing the owning object to be garbage + collected once all other references are dropped. + + Args: + event_type: The :class:`EventType` the handler was registered for. + handler: The callable to remove. + """ + handlers = self._subscriptions.get(event_type) + if handlers is None: + return + with contextlib.suppress(ValueError): + handlers.remove(handler) + __all__ = ["LoggingEventBus"] diff --git a/src/cleveragents/infrastructure/events/protocol.py b/src/cleveragents/infrastructure/events/protocol.py index 2c3cae85f..a8f59fe81 100644 --- a/src/cleveragents/infrastructure/events/protocol.py +++ b/src/cleveragents/infrastructure/events/protocol.py @@ -49,5 +49,23 @@ class EventBus(Protocol): """ ... + def unsubscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> None: + """Remove a previously registered handler for *event_type*. + + If *handler* was not registered for *event_type*, this is a no-op + (no exception is raised). This allows callers to safely call + ``unsubscribe`` during component teardown without tracking whether + ``subscribe`` was ever called. + + Args: + event_type: The :class:`EventType` the handler was registered for. + handler: The callable to remove. + """ + ... + __all__ = ["EventBus"] diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index d55a2eb50..e408f3f72 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -163,6 +163,31 @@ class ReactiveEventBus: raise TypeError("handler must be callable") self._subscriptions.setdefault(event_type, []).append(handler) + def unsubscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> None: + """Remove *handler* from the list of subscribers for *event_type*. + + If *handler* was not registered for *event_type*, this is a no-op — + no exception is raised. This makes it safe to call during component + teardown without tracking whether :meth:`subscribe` was ever called. + + Removing a handler releases the strong reference held by + ``_subscriptions``, allowing the owning object to be garbage + collected once all other references are dropped. + + Args: + event_type: The :class:`EventType` the handler was registered for. + handler: The callable to remove. + """ + handlers = self._subscriptions.get(event_type) + if handlers is None: + return + with contextlib.suppress(ValueError): + handlers.remove(handler) + @property def stream(self) -> Observable: """Read-only observable stream for advanced RxPY operators.