Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 11de2bddb2 | |||
| 20a49d92fd | |||
| 7faab43918 | |||
| 73832cd6bb | |||
| 1e9620223f |
@@ -3,8 +3,6 @@ name: CI
|
||||
on:
|
||||
push:
|
||||
branches: [master, develop]
|
||||
pull_request:
|
||||
branches: [master, develop]
|
||||
|
||||
vars:
|
||||
docker_prefix: "http://harbor.cleverthis.com/docker/"
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -118,6 +118,13 @@ class _FailingEventBus:
|
||||
) -> None:
|
||||
pass
|
||||
|
||||
def unsubscribe(
|
||||
self,
|
||||
event_type: EventType,
|
||||
handler: Any,
|
||||
) -> None:
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Given steps
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user