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
33 changes: 30 additions & 3 deletions src/omnibase_infra/handlers/handler_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,7 @@

from __future__ import annotations

import asyncio
import logging
from uuid import UUID, uuid4

Expand Down Expand Up @@ -126,6 +127,11 @@
)
_SUPPORTED_OPERATIONS: frozenset[str] = frozenset({"db.query", "db.execute"})

# Per-handler pool close timeout (OMN-882)
# asyncpg pool.close() can take 30s timeout x pool_size connections worst case.
# This timeout prevents unbounded shutdown waits.
_POOL_CLOSE_TIMEOUT_SECONDS: float = 5.0
Comment on lines +130 to +133

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Reading handler_db.py around shutdown method ==="
head -n 450 src/omnibase_infra/handlers/handler_db.py | tail -n 50

echo
echo "=== Searching for wait_for and CancelledError handling in handler_db.py ==="
rg -n 'wait_for|CancelledError|except.*Error' src/omnibase_infra/handlers/handler_db.py | head -30

echo
echo "=== Looking for handler shutdown timeout orchestration ==="
fd -type f -name '*.py' -path '*/runtime/*' | xargs rg -l 'handler.*shutdown|shutdown.*timeout' 2>/dev/null | head -5

Repository: OmniNode-ai/omnibase_infra

Length of output: 4161


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Inspecting protocol_lifecycle_executor.py for handler shutdown ==="
fd -type f -name 'protocol_lifecycle_executor.py' -o -name '*lifecycle*executor*'

Repository: OmniNode-ai/omnibase_infra

Length of output: 328


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Full view of handler_db.py shutdown method ==="
wc -l src/omnibase_infra/handlers/handler_db.py
echo
rg -n -A 20 'async def shutdown' src/omnibase_infra/handlers/handler_db.py

Repository: OmniNode-ai/omnibase_infra

Length of output: 1209


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Search for calls to handler shutdown() ==="
rg -n 'shutdown\(\)' src/omnibase_infra --type py | grep -i handler | head -20

echo
echo "=== Search for lifecycle/orchestrator files ==="
fd -name '*orchestr*' -o -name '*lifecycle*' -o -name '*executor*' src/omnibase_infra | head -10

echo
echo "=== Search for task timeout management ==="
rg -n 'wait_for.*shutdown|gather.*shutdown|timeout.*handler' src/omnibase_infra --type py | head -20

Repository: OmniNode-ai/omnibase_infra

Length of output: 2699


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Find all Python files in runtime directory ==="
find src/omnibase_infra/runtime -name '*.py' 2>/dev/null

echo
echo "=== Search for handler shutdown orchestration patterns ==="
rg -n 'handler.*shutdown|shutdown.*handler' src/omnibase_infra --type py -A 2 | head -40

Repository: OmniNode-ai/omnibase_infra

Length of output: 14894


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Read protocol_lifecycle_executor.py around line 307 ==="
sed -n '290,330p' src/omnibase_infra/runtime/protocol_lifecycle_executor.py

echo
echo "=== Search for effective_timeout calculation ==="
rg -n 'effective_timeout' src/omnibase_infra/runtime/protocol_lifecycle_executor.py -B 5 -A 3

Repository: OmniNode-ai/omnibase_infra

Length of output: 4778


🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "=== Check if asyncio.CancelledError is caught anywhere in shutdown path ==="
rg -n 'CancelledError' src/omnibase_infra/handlers/handler_db.py

echo
echo "=== Verify asyncio.wait_for behavior with nested contexts ==="
python3 << 'PYTHON_EOF'
import asyncio
import inspect

# Check wait_for signature and CancelledError behavior
print("asyncio.wait_for behavior with nested timeouts:")
print("- Outer wait_for timeout < inner timeout raises asyncio.TimeoutError (wrapped as CancelledError)")
print("- Caller of wait_for must handle TimeoutError explicitly")
print("- CancelledError is a BaseException, NOT caught by 'except TimeoutError'")
print()
print("Code path if outer timeout fires before inner timeout:")
print("1. Outer wait_for() times out -> raises asyncio.TimeoutError")
print("2. Inner except TimeoutError does NOT catch this")
print("3. Finally block executes, clearing self._pool without terminate()")
print("4. Leaked connections remain active")
PYTHON_EOF

Repository: OmniNode-ai/omnibase_infra

Length of output: 145


Handle cancellation path in pool close; without CancelledError handling, connections leak on cancellation.

The outer handler.shutdown() call at protocol_lifecycle_executor.py:307 wraps the handler in asyncio.wait_for(handler.shutdown(), timeout=effective_timeout). If this outer timeout or task cancellation fires before the inner pool close completes, a CancelledError is raised. Since CancelledError doesn't inherit from TimeoutError, the exception bypasses the handler at line 424, but the finally block at line 435 still executes and clears self._pool without calling terminate(). This leaves live database connections active.

Add except asyncio.CancelledError handling to ensure terminate() is called:

Suggested fix
         if self._pool is not None:
             try:
                 await asyncio.wait_for(
                     self._pool.close(), timeout=_POOL_CLOSE_TIMEOUT_SECONDS
                 )
             except TimeoutError:
                 logger.warning(
                     "Database pool close timed out, terminating connections",
                     extra={
                         "timeout_seconds": _POOL_CLOSE_TIMEOUT_SECONDS,
                         "pool_size": self._pool_size,
                     },
                 )
                 # Terminate remaining connections forcefully
                 self._pool.terminate()
+            except asyncio.CancelledError:
+                logger.warning(
+                    "Database pool close cancelled, terminating connections",
+                    extra={"pool_size": self._pool_size},
+                )
+                self._pool.terminate()
+                raise
             finally:
                 self._pool = None
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@src/omnibase_infra/handlers/handler_db.py` around lines 130 - 133, The pool
close path can leak connections if a CancelledError escapes before pool
termination; update the shutdown routine in handler.shutdown() (the block that
closes self._pool and calls terminate()) to add an except asyncio.CancelledError
clause that calls self._pool.terminate() (or the handler's terminate() method)
to forcibly close connections, then re-raises the CancelledError so cancellation
propagates; ensure the existing finally that clears self._pool still runs after
termination so no live pool reference is left.


# Error message prefixes for PostgreSQL errors
# Used by _map_postgres_error to build descriptive error messages
_POSTGRES_ERROR_PREFIXES: dict[type[asyncpg.PostgresError], str] = {
Expand Down Expand Up @@ -396,16 +402,37 @@ async def initialize(self, config: dict[str, object]) -> None:
) from e

async def shutdown(self) -> None:
"""Close database connection pool and release resources."""
"""Close database connection pool and release resources.

The pool close is wrapped with a timeout to prevent unbounded shutdown
waits. asyncpg pool.close() can take up to 30s per connection x pool_size
in the worst case (e.g., 150s for 5 connections). The timeout ensures
predictable shutdown behavior. If the pool close times out, the pool
reference is still cleared to allow garbage collection.
"""
# Reset circuit breaker state
if self._circuit_breaker_initialized:
async with self._circuit_breaker_lock:
await self._reset_circuit_breaker()
self._circuit_breaker_initialized = False

if self._pool is not None:
await self._pool.close()
self._pool = None
try:
await asyncio.wait_for(
self._pool.close(), timeout=_POOL_CLOSE_TIMEOUT_SECONDS
)
except TimeoutError:
logger.warning(
"Database pool close timed out, terminating connections",
extra={
"timeout_seconds": _POOL_CLOSE_TIMEOUT_SECONDS,
"pool_size": self._pool_size,
},
)
# Terminate remaining connections forcefully
self._pool.terminate()
finally:
self._pool = None
self._initialized = False
logger.info("HandlerDb shutdown complete")

Expand Down
24 changes: 21 additions & 3 deletions src/omnibase_infra/handlers/handler_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

from __future__ import annotations

import asyncio
import json
import logging
from uuid import UUID, uuid4
Expand Down Expand Up @@ -75,6 +76,9 @@
# Streaming chunk size for responses without Content-Length header
_STREAMING_CHUNK_SIZE: int = 8192 # 8 KB chunks

# Per-handler client close timeout (OMN-882)
_CLIENT_CLOSE_TIMEOUT_SECONDS: float = 5.0

# Handler ID for ModelHandlerOutput
HANDLER_ID_HTTP: str = "http-handler"

Expand Down Expand Up @@ -296,10 +300,24 @@ async def initialize(self, config: dict[str, object]) -> None:
) from e

async def shutdown(self) -> None:
"""Close HTTP client and release resources."""
"""Close HTTP client and release resources.

The client close is wrapped with a timeout to prevent unbounded shutdown
waits if the underlying connection pool has hung connections.
"""
if self._client is not None:
await self._client.aclose()
self._client = None
try:
await asyncio.wait_for(
self._client.aclose(),
timeout=_CLIENT_CLOSE_TIMEOUT_SECONDS,
)
except TimeoutError:
logger.warning(
"HTTP client close timed out",
extra={"timeout_seconds": _CLIENT_CLOSE_TIMEOUT_SECONDS},
)
finally:
self._client = None
self._initialized = False
logger.info("HandlerHttpRest shutdown complete")
Comment thread
coderabbitai[bot] marked this conversation as resolved.

Expand Down
39 changes: 27 additions & 12 deletions src/omnibase_infra/runtime/models/model_shutdown_config.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,6 @@
# SPDX-License-Identifier: MIT
# Copyright (c) 2025 OmniNode Team
"""Shutdown Configuration Model.

The Pydantic model for shutdown configuration.
"""
"""Shutdown configuration for ONEX runtime graceful shutdown."""

from __future__ import annotations

Expand Down Expand Up @@ -56,23 +53,30 @@ def _coerce_grace_period(v: object) -> int:
class ModelShutdownConfig(BaseModel):
"""Shutdown configuration model.

Defines graceful shutdown parameters.
Defines graceful shutdown parameters including per-handler timeout controls.

Attributes:
grace_period_seconds: Time in seconds to wait for graceful shutdown.
grace_period_seconds: Time in seconds to wait for overall graceful shutdown.
Must be >= 0. A value of 0 means immediate shutdown with no grace
period (use with caution as in-flight operations may be interrupted).
handler_shutdown_timeout_seconds: Maximum time in seconds to wait for each
individual handler's shutdown method to complete. If a handler exceeds
this timeout, the shutdown is considered failed for that handler but
other handlers continue their shutdown. This prevents a single slow
handler (e.g., asyncpg pool close with 30s per-connection timeout)
from blocking the entire shutdown sequence.

Edge Cases:
- 0: Immediate shutdown, no waiting for in-flight operations
- Values > 3600: Rejected by Pydantic validation (le=3600 constraint);
consider using reasonable timeouts (30-120 seconds recommended)
- grace_period_seconds=0: Immediate shutdown, no waiting for in-flight operations
- Values > 3600: Rejected by Pydantic validation (le=3600 constraint)
- Negative values: Rejected by Pydantic validation (ge=0 constraint)
- handler_shutdown_timeout_seconds=0: Not allowed (ge=1 constraint)

Production Recommendation:
Set grace_period_seconds between 30-120 seconds for production deployments
to allow sufficient time for in-flight operations while preventing excessive
delays during shutdown sequences.
Set grace_period_seconds between 30-120 seconds for production deployments.
Set handler_shutdown_timeout_seconds to 10-15 seconds to prevent any single
handler from consuming the entire grace period. The handler timeout should
always be less than the overall grace period.
"""

model_config = ConfigDict(
Expand All @@ -90,5 +94,16 @@ class ModelShutdownConfig(BaseModel):
description="Time in seconds to wait for graceful shutdown (0-3600)",
)

handler_shutdown_timeout_seconds: float = Field(
default=10.0,
ge=1.0,
le=300.0, # Max 5 minutes per handler
strict=False,
description=(
"Maximum time in seconds to wait for each individual handler shutdown. "
"Prevents a single slow handler from blocking the entire shutdown sequence."
),
)


__all__: list[str] = ["ModelShutdownConfig"]
114 changes: 95 additions & 19 deletions src/omnibase_infra/runtime/protocol_lifecycle_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,13 +2,9 @@
# Copyright (c) 2025 OmniNode Team
"""Protocol lifecycle executor for ONEX Infrastructure.

The ProtocolLifecycleExecutor class for managing protocol
handler lifecycle operations including shutdown, health checks, and priority
ordering.

The ProtocolLifecycleExecutor is a reusable utility extracted from
RuntimeHostProcess to enable lifecycle management in other contexts (e.g., test
fixtures, standalone management, future node implementations).
Reusable utility extracted from RuntimeHostProcess to manage protocol handler
lifecycle operations (shutdown, health checks, priority ordering) in multiple
contexts: runtime host, test fixtures, standalone management.

Example Usage:
```python
Expand Down Expand Up @@ -58,17 +54,21 @@
MAX_HEALTH_CHECK_TIMEOUT = 60.0
DEFAULT_HEALTH_CHECK_TIMEOUT = 5.0

# Handler shutdown timeout bounds (per OMN-882)
MIN_HANDLER_SHUTDOWN_TIMEOUT = 1.0
MAX_HANDLER_SHUTDOWN_TIMEOUT = 300.0
DEFAULT_HANDLER_SHUTDOWN_TIMEOUT = 10.0


class ProtocolLifecycleExecutor:
"""Executes protocol lifecycle operations (shutdown, health check, priority).

Reusable utility for managing protocol lifecycles. Used by
RuntimeHostProcess and can be used by future node implementations.

Exports:
- Priority-based shutdown ordering (higher priority handlers shutdown first)
- Health check execution with configurable timeouts
- Parallel shutdown within priority groups for performance
Capabilities: priority-based shutdown ordering (higher priority first),
health check execution with configurable timeouts, parallel shutdown
within priority groups.

Thread Safety:
This class is thread-safe for concurrent method calls.
Expand Down Expand Up @@ -106,6 +106,7 @@ class ProtocolLifecycleExecutor:

Attributes:
health_check_timeout_seconds: Default timeout for health checks (1-60 seconds).
handler_shutdown_timeout_seconds: Per-handler shutdown timeout (1-300 seconds).

Example:
```python
Expand All @@ -129,16 +130,24 @@ class ProtocolLifecycleExecutor:
"""

def __init__(
self, health_check_timeout_seconds: float = DEFAULT_HEALTH_CHECK_TIMEOUT
self,
health_check_timeout_seconds: float = DEFAULT_HEALTH_CHECK_TIMEOUT,
handler_shutdown_timeout_seconds: float = DEFAULT_HANDLER_SHUTDOWN_TIMEOUT,
) -> None:
"""Initialize the protocol lifecycle executor.

Args:
health_check_timeout_seconds: Default timeout for health checks.
Valid range: 1-60 seconds per ModelLifecycleSubcontract.
Values outside this range are clamped with a warning.
handler_shutdown_timeout_seconds: Per-handler shutdown timeout.
Valid range: 1-300 seconds. Each individual handler's shutdown()
call is wrapped with asyncio.wait_for() using this timeout.
Prevents a single slow handler (e.g., asyncpg pool close) from
blocking the entire shutdown sequence.
Values outside this range are clamped with a warning.
"""
# Validate and clamp timeout to valid bounds
# Validate and clamp health check timeout to valid bounds
if (
health_check_timeout_seconds < MIN_HEALTH_CHECK_TIMEOUT
or health_check_timeout_seconds > MAX_HEALTH_CHECK_TIMEOUT
Expand All @@ -160,7 +169,33 @@ def __init__(
min(health_check_timeout_seconds, MAX_HEALTH_CHECK_TIMEOUT),
)

# Validate and clamp handler shutdown timeout to valid bounds
if (
handler_shutdown_timeout_seconds < MIN_HANDLER_SHUTDOWN_TIMEOUT
or handler_shutdown_timeout_seconds > MAX_HANDLER_SHUTDOWN_TIMEOUT
):
logger.warning(
"handler_shutdown_timeout_seconds out of valid range, clamping",
extra={
"original_value": handler_shutdown_timeout_seconds,
"min_value": MIN_HANDLER_SHUTDOWN_TIMEOUT,
"max_value": MAX_HANDLER_SHUTDOWN_TIMEOUT,
"clamped_value": max(
MIN_HANDLER_SHUTDOWN_TIMEOUT,
min(
handler_shutdown_timeout_seconds,
MAX_HANDLER_SHUTDOWN_TIMEOUT,
),
),
},
)
handler_shutdown_timeout_seconds = max(
MIN_HANDLER_SHUTDOWN_TIMEOUT,
min(handler_shutdown_timeout_seconds, MAX_HANDLER_SHUTDOWN_TIMEOUT),
)

self._health_check_timeout_seconds: float = health_check_timeout_seconds
self._handler_shutdown_timeout_seconds: float = handler_shutdown_timeout_seconds

@property
def health_check_timeout_seconds(self) -> float:
Expand All @@ -171,6 +206,15 @@ def health_check_timeout_seconds(self) -> float:
"""
return self._health_check_timeout_seconds

@property
def handler_shutdown_timeout_seconds(self) -> float:
"""Return the configured per-handler shutdown timeout.

Returns:
The timeout in seconds for individual handler shutdowns.
"""
return self._handler_shutdown_timeout_seconds

@staticmethod
def get_shutdown_priority(handler: ProtocolContainerAware) -> int:
"""Get shutdown priority for a handler.
Expand Down Expand Up @@ -226,26 +270,42 @@ async def shutdown_handler(
self,
handler_type: str,
handler: ProtocolContainerAware,
timeout_seconds: float = -1.0,
) -> ModelLifecycleResult:
"""Shutdown a single handler with error handling.
"""Shutdown a single handler with per-handler timeout and error handling.

This method performs individual handler shutdown with comprehensive error
handling to ensure all handlers get a chance to cleanup even if one fails.
This method performs individual handler shutdown wrapped with
asyncio.wait_for() to enforce a per-handler timeout. This prevents a
single slow handler (e.g., asyncpg pool close with 30s per-connection
timeout x 5 connections = 150s worst case) from blocking the entire
shutdown sequence.

Args:
handler_type: The handler type identifier.
handler: The handler instance to shutdown.
timeout_seconds: Override timeout for this specific shutdown. If
negative (default: -1.0), uses the configured
handler_shutdown_timeout_seconds.

Returns:
ModelLifecycleResult with handler_type, success status, and optional
error_message if the shutdown failed.
error_message if the shutdown failed or timed out.
"""
effective_timeout = (
timeout_seconds
if timeout_seconds > 0
else self._handler_shutdown_timeout_seconds
)

try:
if hasattr(handler, "shutdown"):
await handler.shutdown()
await asyncio.wait_for(handler.shutdown(), timeout=effective_timeout)
logger.debug(
"Handler shutdown completed",
extra={"handler_type": handler_type},
extra={
"handler_type": handler_type,
"timeout_seconds": effective_timeout,
},
)
return ModelLifecycleResult.succeeded(handler_type)
else:
Expand All @@ -255,6 +315,19 @@ async def shutdown_handler(
extra={"handler_type": handler_type},
)
return ModelLifecycleResult.succeeded(handler_type)
except TimeoutError:
# Handler shutdown exceeded the per-handler timeout
logger.warning(
"Handler shutdown timed out",
extra={
"handler_type": handler_type,
"timeout_seconds": effective_timeout,
},
)
return ModelLifecycleResult.failed(
handler_type,
f"Shutdown timed out after {effective_timeout}s",
)
except Exception as e:
# Log exception but return failure status instead of raising
# This ensures all handlers get a chance to cleanup even if one fails
Expand Down Expand Up @@ -428,8 +501,11 @@ async def shutdown_handlers_by_priority(


__all__: list[str] = [
"DEFAULT_HANDLER_SHUTDOWN_TIMEOUT",
"DEFAULT_HEALTH_CHECK_TIMEOUT",
"MAX_HANDLER_SHUTDOWN_TIMEOUT",
"MAX_HEALTH_CHECK_TIMEOUT",
"MIN_HANDLER_SHUTDOWN_TIMEOUT",
"MIN_HEALTH_CHECK_TIMEOUT",
"ProtocolLifecycleExecutor",
]
Loading
Loading