Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 60 additions & 3 deletions src/omnibase_core/mixins/mixin_node_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,12 @@ class MixinNodeService:
- Emit TOOL_RESPONSE events with results
- Provide health monitoring and graceful shutdown
- Support asyncio event loop for concurrent operations

Thread Safety:
All service instances must run within the same asyncio event loop.
The _shutdown_event (asyncio.Event) is only safe for coordination within
a single event loop. For multi-threaded scenarios with separate event
loops, use threading.Event instead.
"""
Comment thread
coderabbitai[bot] marked this conversation as resolved.

# Type annotations for attributes set via object.__setattr__()
Expand All @@ -82,6 +88,7 @@ class MixinNodeService:
_start_time: float | None
_shutdown_requested: bool
_shutdown_callbacks: list[Callable[[], None]]
_shutdown_event: asyncio.Event | None

def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initialize the service mixin."""
Expand All @@ -104,6 +111,8 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
# Shutdown handling
object.__setattr__(self, "_shutdown_requested", False)
object.__setattr__(self, "_shutdown_callbacks", [])
# Shutdown event for immediate task cancellation (created lazily in start_service_mode)
object.__setattr__(self, "_shutdown_event", None)

async def start_service_mode(self) -> None:
"""
Expand All @@ -123,6 +132,11 @@ async def start_service_mode(self) -> None:
self._service_running = True
self._start_time = time.time()

# Create shutdown event for immediate task cancellation
# This allows health monitor and service loops to wake up immediately
# when shutdown is requested instead of waiting for sleep to complete
self._shutdown_event = asyncio.Event()

# Publish introspection for service discovery
self._publish_introspection_event()

Expand Down Expand Up @@ -161,6 +175,12 @@ async def stop_service_mode(self) -> None:
self._log_info("Stopping service mode...")
self._shutdown_requested = True

# Signal shutdown event to wake up any sleeping tasks immediately
# This allows health monitor loop to respond to cancellation without
# waiting for the full sleep interval (30 seconds)
if self._shutdown_event is not None:
self._shutdown_event.set()

try:
# Emit shutdown event
await self._emit_shutdown_event()
Expand Down Expand Up @@ -197,6 +217,10 @@ async def aclose(self) -> None:
especially the health monitor task. Can be called directly by tests
or used as an async context manager exit.
"""
# Signal shutdown event to wake up any sleeping tasks immediately
if self._shutdown_event is not None:
self._shutdown_event.set()

# Cancel health monitoring task if it exists
await self._cleanup_health_task()

Expand Down Expand Up @@ -377,7 +401,17 @@ async def _service_event_loop(self) -> None:
try:
while self._service_running and not self._shutdown_requested:
# Process any pending events (depending on event bus implementation)
await asyncio.sleep(0.1) # Small delay to prevent busy waiting
# Use shutdown event for interruptible sleep to allow immediate shutdown
if self._shutdown_event is not None:
try:
await asyncio.wait_for(self._shutdown_event.wait(), timeout=0.1)
# If we get here, shutdown was signaled
break
except TimeoutError:
# Normal timeout - continue event loop
pass
else:
await asyncio.sleep(0.1) # Small delay to prevent busy waiting

except asyncio.CancelledError:
# CRITICAL: Do not log here - file handles may be closed during teardown
Expand All @@ -400,8 +434,21 @@ async def _health_monitor_loop(self) -> None:
f"Health: {health['active_invocations']} active, {health['success_rate']:.2%} success rate",
)

# Wait before next health check
await asyncio.sleep(30)
# Wait before next health check, but allow immediate wakeup on shutdown
# This prevents the "Task was destroyed but it is pending!" warning
# by responding immediately to shutdown signals instead of sleeping
if self._shutdown_event is not None:
try:
# Wait for shutdown event with timeout (health check interval)
await asyncio.wait_for(self._shutdown_event.wait(), timeout=30)
# If we get here, shutdown was signaled
break
except TimeoutError:
# Normal timeout - continue health monitoring
pass
else:
# Fallback to simple sleep if no shutdown event (shouldn't happen)
await asyncio.sleep(30)

except asyncio.CancelledError:
# CRITICAL: Do not log here - file handles may be closed during teardown
Expand Down Expand Up @@ -651,6 +698,16 @@ def signal_handler(signum: int, _frame: Any) -> None:
f"Received signal {signum}, initiating graceful shutdown",
)
self._shutdown_requested = True
# Signal shutdown event to wake up any sleeping tasks immediately
# Use call_soon_threadsafe since signal handlers run in main thread
# but asyncio.Event.set() should be called from the event loop thread
if self._shutdown_event is not None:
try:
loop = asyncio.get_running_loop()
loop.call_soon_threadsafe(self._shutdown_event.set)
except RuntimeError:
# No running loop - set directly (may be during shutdown)
self._shutdown_event.set()

signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)
Expand Down
32 changes: 32 additions & 0 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,38 @@
- "coroutine was never awaited" warnings are NOT suppressed (indicate real bugs)

See pyproject.toml filterwarnings configuration for details.

Global Patching Side Effects:
-----------------------------
WARNING: This conftest applies GLOBAL patches to asyncio exception handling.

The `event_loop_cleanup` and `cleanup_service_tasks` fixtures temporarily set
`loop.set_exception_handler(lambda loop, context: None)` during teardown. This
affects ALL tests in the suite when these fixtures are active.

What is patched:
- The asyncio event loop's exception handler is set to a no-op lambda
- This occurs during the teardown phase of each test
- The original handler is restored after cleanup completes

Side Effects:
- All asyncio exceptions during test teardown are silently swallowed
- "Task was destroyed but it is pending!" warnings are suppressed (intentional)
- Any CancelledError or other exceptions raised during task cancellation are hidden
- Tests cannot rely on the exception handler being called during teardown
- If cleanup fails, no error messages will be logged

Why This Exists:
- Health monitor tasks and service loops may have long sleep intervals
- Cancelling these tasks during teardown generates noisy but harmless warnings
- These warnings obscure real test failures in CI output
- The patching is scoped to teardown only, not during test execution

Mitigation:
- Tests that need to verify exception handler behavior should mock it explicitly
- Use explicit task cancellation and await BEFORE test teardown if verification needed
- Consider using `pytest.mark.filterwarnings` for test-specific exception handling
- The `mock_event_loop` fixture provides isolated event loop testing without global effects
"""

import asyncio
Expand Down
Loading