From 010eeb034fd86a53624f5931bca95de6edaf76c3 Mon Sep 17 00:00:00 2001 From: "cristiano.hoshikawa" Date: Fri, 7 Aug 2026 21:28:54 -0300 Subject: [PATCH] bug fix: Deadlock Sequence on Pubsub --- .../FIX_DEADLOCK_SEQUENCE_CROSS_LOOP.md | 37 ++++++++ .../agent_framework/analytics/tim_sequence.py | 43 +++++---- .../src/agent_framework/observer.py | 78 +++++++++++++++- .../test_observer_cross_loop_deadlock_fix.py | 93 +++++++++++++++++++ 4 files changed, 232 insertions(+), 19 deletions(-) create mode 100644 Documentacao/FIX_DEADLOCK_SEQUENCE_CROSS_LOOP.md create mode 100644 tests/test_observer_cross_loop_deadlock_fix.py diff --git a/Documentacao/FIX_DEADLOCK_SEQUENCE_CROSS_LOOP.md b/Documentacao/FIX_DEADLOCK_SEQUENCE_CROSS_LOOP.md new file mode 100644 index 0000000..b44a37c --- /dev/null +++ b/Documentacao/FIX_DEADLOCK_SEQUENCE_CROSS_LOOP.md @@ -0,0 +1,37 @@ +# Correção: deadlock/espera cross-loop na geração de sequence + +## Problema + +A API síncrona `agent_framework.observer.event()` podia ser chamada em uma worker thread sem event loop ativo. Nesse caso, a implementação anterior executava `asyncio.run(aevent(...))`, criando um novo event loop temporário. Ao mesmo tempo, `analytics/tim_sequence.py` compartilhava instâncias globais de `asyncio.Lock` (`_mongo_index_lock` e `_memory_lock`) entre chamadas que podiam vir de event loops diferentes. + +Na primeira operação Mongo, `_ensure_mongo_ttl_index_once()` mantinha `_mongo_index_lock` durante a criação do índice TTL. A contenção por outro loop podia deixar a segunda chamada aguardando indefinidamente. + +## Alterações aplicadas + +1. `observer.py` + - removido `asyncio.run()` do caminho síncrono de `event()`; + - adicionado um event loop dedicado e reutilizável para chamadas síncronas; + - submissão cross-thread feita com `asyncio.run_coroutine_threadsafe()`; + - encerramento best-effort do loop no shutdown do processo. + +2. `analytics/tim_sequence.py` + - `_mongo_index_lock`: `asyncio.Lock` -> `threading.Lock`; + - `_memory_lock`: `asyncio.Lock` -> `threading.Lock`; + - inicialização do índice TTL movida para uma função síncrona protegida por lock de thread e chamada via `asyncio.to_thread()`; + - o contador de fallback em memória usa uma seção crítica curta e thread-safe. + +3. Testes + - `tests/test_observer_cross_loop_deadlock_fix.py` valida: + - múltiplas worker threads usando `event()` compartilham o mesmo loop síncrono do observer; + - sequence em memória permanece monotônica entre event loops independentes; + - criação do índice TTL ocorre apenas uma vez sob contenção cross-loop. + +## Validação executada + +```bash +PYTHONPATH=libs/agent_framework/src pytest -q tests/test_observer_cross_loop_deadlock_fix.py +``` + +Resultado: `3 passed`. + +A suíte completa do repositório possui falhas preexistentes/independentes desta alteração, incluindo conflitos de coleta de arquivos `test_long_term_memory.py`, caminhos estáticos de template e testes de checkpoint/workflow. Esses itens não foram alterados por esta correção. diff --git a/libs/agent_framework/src/agent_framework/analytics/tim_sequence.py b/libs/agent_framework/src/agent_framework/analytics/tim_sequence.py index 52945a2..388e17b 100644 --- a/libs/agent_framework/src/agent_framework/analytics/tim_sequence.py +++ b/libs/agent_framework/src/agent_framework/analytics/tim_sequence.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import logging import os +import threading from collections import defaultdict from datetime import datetime, timedelta, timezone from typing import Any, Literal @@ -12,7 +13,7 @@ logger = logging.getLogger("agent_framework.analytics.tim_sequence") # In-process fallback. This is not cross-process/global, but keeps telemetry alive # when the configured shared sequence backend is unavailable, matching the # framework principle that observability must not break business execution. -_memory_lock = asyncio.Lock() +_memory_lock = threading.Lock() _memory_counters: dict[str, int] = defaultdict(int) SequenceProvider = Literal["auto", "redis", "mongodb", "mongo", "memory", "none"] @@ -152,7 +153,7 @@ async def _next_sequence_redis(key: str, ttl_seconds: int) -> int | None: _mongo_index_checked = False -_mongo_index_lock = asyncio.Lock() +_mongo_index_lock = threading.Lock() def _next_sequence_mongodb_sync( @@ -204,37 +205,43 @@ def _next_sequence_mongodb_sync( client.close() -async def _ensure_mongo_ttl_index_once(ttl_seconds: int) -> None: - """Best-effort TTL index creation for Mongo sequence docs. +def _ensure_mongo_ttl_index_once_sync(ttl_seconds: int) -> None: + """Best-effort TTL index initialization, safe across threads/event loops. - The sequence still works without this index. If the application user lacks - index privileges, we only log and continue. + ``asyncio.Lock`` must not be shared by independent event loops. Observer + compatibility calls may originate in worker threads, so this one-time + process-local guard deliberately uses ``threading.Lock``. The blocking + Mongo operation is executed by the async wrapper in a worker thread. """ global _mongo_index_checked if _mongo_index_checked or ttl_seconds <= 0 or not _mongo_uri(): return - async with _mongo_index_lock: + with _mongo_index_lock: if _mongo_index_checked: return try: from pymongo import MongoClient # type: ignore - def _create() -> None: - client = MongoClient(_mongo_uri()) - try: - collection = client[_mongo_database()][_mongo_collection()] - collection.create_index("expiresAt", expireAfterSeconds=0, background=True) - finally: - client.close() - - await asyncio.to_thread(_create) + client = MongoClient(_mongo_uri()) + try: + collection = client[_mongo_database()][_mongo_collection()] + collection.create_index("expiresAt", expireAfterSeconds=0, background=True) + finally: + client.close() except Exception: logger.warning("tim_sequence.mongodb_ttl_index_failed", exc_info=True) finally: + # The index is an observability housekeeping concern, not a + # prerequisite for sequence generation. Do not retry on every + # event if the application user lacks index privileges. _mongo_index_checked = True +async def _ensure_mongo_ttl_index_once(ttl_seconds: int) -> None: + await asyncio.to_thread(_ensure_mongo_ttl_index_once_sync, ttl_seconds) + + async def _next_sequence_mongodb( key: str, agent_id: str | None, @@ -260,7 +267,9 @@ async def _next_sequence_mongodb( async def _next_sequence_memory(key: str) -> int: - async with _memory_lock: + # Tiny in-process critical section; a thread lock is intentional because + # this fallback can be reached from more than one asyncio event loop. + with _memory_lock: _memory_counters[key] += 1 return _memory_counters[key] diff --git a/libs/agent_framework/src/agent_framework/observer.py b/libs/agent_framework/src/agent_framework/observer.py index eb28d7d..c2a3db7 100644 --- a/libs/agent_framework/src/agent_framework/observer.py +++ b/libs/agent_framework/src/agent_framework/observer.py @@ -14,9 +14,10 @@ rails, bridges e comandos de negócio sem quebrar o turno do cliente. """ import asyncio +import atexit import logging import os -from threading import Lock +from threading import Event, Lock, Thread from typing import Any from agent_framework.analytics.factory import create_analytics_publisher @@ -29,6 +30,74 @@ _GLOBAL_CONFIG: dict[str, Any] = {} _LOCK = Lock() +class _SyncEventLoopBridge: + """Own one reusable event loop for synchronous observer calls. + + The legacy ``event()`` API is frequently invoked from worker threads that + do not own an asyncio loop. Creating a fresh loop with ``asyncio.run()`` + for every such call makes the same global observer reachable from multiple + temporary loops. This bridge keeps those synchronous calls on one stable + loop and submits work through asyncio's thread-safe API. + """ + + def __init__(self) -> None: + self._start_lock = Lock() + self._ready = Event() + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: Thread | None = None + + def _thread_main(self) -> None: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + self._loop = loop + self._ready.set() + try: + loop.run_forever() + finally: + pending = asyncio.all_tasks(loop) + for task in pending: + task.cancel() + if pending: + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + loop.close() + + def _ensure_started(self) -> asyncio.AbstractEventLoop: + loop = self._loop + if loop is not None and loop.is_running(): + return loop + with self._start_lock: + loop = self._loop + if loop is None or not loop.is_running(): + self._ready.clear() + self._thread = Thread( + target=self._thread_main, + name="agent-framework-observer-loop", + daemon=True, + ) + self._thread.start() + self._ready.wait() + assert self._loop is not None + return self._loop + + def run(self, coro: Any) -> Any: + loop = self._ensure_started() + future = asyncio.run_coroutine_threadsafe(coro, loop) + return future.result() + + def close(self) -> None: + loop = self._loop + thread = self._thread + if loop is None or not loop.is_running(): + return + loop.call_soon_threadsafe(loop.stop) + if thread is not None and thread.is_alive(): + thread.join(timeout=2.0) + + +_SYNC_EVENT_LOOP = _SyncEventLoopBridge() +atexit.register(_SYNC_EVENT_LOOP.close) + + def _truthy(value: Any, default: bool = False) -> bool: if value is None: return default @@ -202,7 +271,12 @@ def event( try: loop = asyncio.get_running_loop() except RuntimeError: - return asyncio.run(aevent(name, data=data, metadata=metadata, event_test=event_test)) + # Do not create a temporary event loop in every worker thread. Route + # synchronous compatibility calls to one stable observer loop using + # asyncio's thread-safe submission primitive. + return _SYNC_EVENT_LOOP.run( + aevent(name, data=data, metadata=metadata, event_test=event_test) + ) task = loop.create_task(aevent(name, data=data, metadata=metadata, event_test=event_test)) task.add_done_callback(_log_task_exception) diff --git a/tests/test_observer_cross_loop_deadlock_fix.py b/tests/test_observer_cross_loop_deadlock_fix.py new file mode 100644 index 0000000..825a25d --- /dev/null +++ b/tests/test_observer_cross_loop_deadlock_fix.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import asyncio +import sys +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace + +from agent_framework import observer +from agent_framework.analytics import tim_sequence + + +def test_sync_event_calls_share_one_stable_asyncio_loop(monkeypatch): + seen_loop_ids: set[int] = set() + seen_lock = threading.Lock() + + async def fake_aevent(name: str, **kwargs): + loop_id = id(asyncio.get_running_loop()) + with seen_lock: + seen_loop_ids.add(loop_id) + await asyncio.sleep(0.01) + return {"eventType": name, "loop_id": loop_id} + + monkeypatch.setattr(observer, "aevent", fake_aevent) + + with ThreadPoolExecutor(max_workers=8) as pool: + results = list(pool.map(lambda i: observer.event(f"IC.TEST.{i}"), range(24))) + + assert len(seen_loop_ids) == 1 + assert {item["loop_id"] for item in results} == seen_loop_ids + + +def test_memory_sequence_is_safe_across_independent_event_loops(monkeypatch): + monkeypatch.setenv("PUBSUB_SEQUENCE_ENABLED", "true") + monkeypatch.setenv("PUBSUB_SEQUENCE_PROVIDER", "memory") + tim_sequence._memory_counters.clear() + + def one_call(_: int) -> int | None: + return asyncio.run( + tim_sequence.next_sequence( + "agent-a", + "session-a", + "transaction-cross-loop", + ) + ) + + with ThreadPoolExecutor(max_workers=12) as pool: + values = list(pool.map(one_call, range(120))) + + assert sorted(values) == list(range(1, 121)) + + +def test_mongo_ttl_index_guard_is_thread_safe_across_event_loops(monkeypatch): + tim_sequence._mongo_index_checked = False + monkeypatch.setenv("PUBSUB_SEQUENCE_MONGODB_URI", "mongodb://fake") + + calls = 0 + calls_lock = threading.Lock() + + class FakeCollection: + def create_index(self, *args, **kwargs): + nonlocal calls + with calls_lock: + calls += 1 + # Enlarge the contention window that previously exposed the + # cross-event-loop asyncio.Lock issue. + time.sleep(0.05) + + class FakeDatabase: + def __getitem__(self, name): + return FakeCollection() + + class FakeMongoClient: + def __init__(self, uri): + self.uri = uri + + def __getitem__(self, name): + return FakeDatabase() + + def close(self): + return None + + monkeypatch.setitem(sys.modules, "pymongo", SimpleNamespace(MongoClient=FakeMongoClient)) + + def ensure_index(_: int) -> None: + asyncio.run(tim_sequence._ensure_mongo_ttl_index_once(60)) + + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(ensure_index, range(16))) + + assert calls == 1 + assert tim_sequence._mongo_index_checked is True