wehub-resource-sync
197 lines
6.7 KiB
Python
197 lines
6.7 KiB
Python
"""Regression tests for the synchronous-wrapper event-loop guard.
|
|
|
|
``LightRAG``'s synchronous wrappers (``insert``, ``query``,
|
|
``delete_by_entity`` …) all delegate to :func:`lightrag.lightrag._run_sync`,
|
|
which drives the matching ``a*`` coroutine via ``loop.run_until_complete()``.
|
|
|
|
That call is only valid when (a) no event loop is already running on the
|
|
current thread, and (b) the loop it drives is the same one the instance's
|
|
storages were initialized on (``LightRAG._owning_loop``). Two misuse modes
|
|
break this:
|
|
|
|
* Called from within a running loop it raises ``RuntimeError: This event loop
|
|
is already running`` — a fail-fast error, *not* a deadlock.
|
|
* Driven from a different — but still alive — loop, e.g.
|
|
``loop.run_in_executor(None, rag.insert, …)`` runs the wrapper on a pool
|
|
thread that spins up a fresh loop while the app's loop keeps running — binds
|
|
the shared ``asyncio.Lock`` objects in ``lightrag.kg.shared_storage`` to the
|
|
wrong loop (``<Lock> is bound to a different event loop`` / a stall).
|
|
|
|
The cross-loop check only fires while ``owning_loop`` is still open. A *closed*
|
|
owning loop (the ``rag = asyncio.run(initialize_rag())`` pattern, where
|
|
``asyncio.run`` closes the loop after ``initialize_storages()``) is let through
|
|
to run on a fresh loop, preserving long-standing behavior.
|
|
|
|
Both misuse modes have the same fix: use the ``a*`` coroutine directly.
|
|
``_run_sync`` detects the running loop up front and compares the loop it is
|
|
about to drive against ``owning_loop``, raising one clear, actionable error for
|
|
each. These tests pin that behaviour (and guard against regressing back to the
|
|
misleading "deadlock" wording / un-awaited-coroutine leak). See
|
|
HKUDS/LightRAG #1968.
|
|
"""
|
|
|
|
import asyncio
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
import pytest
|
|
|
|
from lightrag.lightrag import _run_sync
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_run_sync_runs_coroutine_when_no_loop_running():
|
|
"""With no running loop, the coroutine runs to completion and returns."""
|
|
|
|
def factory():
|
|
async def _coro():
|
|
return 42
|
|
|
|
return _coro()
|
|
|
|
result = _run_sync(factory, sync_name="insert", async_name="ainsert")
|
|
assert result == 42
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_run_sync_raises_clear_error_inside_running_loop():
|
|
"""Inside a running loop the guard raises a RuntimeError pointing at the
|
|
async alternative — and the message is honest (no "deadlock" claim)."""
|
|
|
|
def factory(): # pragma: no cover - must never be reached
|
|
raise AssertionError("coro_factory must not be called inside a loop")
|
|
|
|
async def _inside_loop():
|
|
return _run_sync(factory, sync_name="insert", async_name="ainsert")
|
|
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
asyncio.run(_inside_loop())
|
|
|
|
message = str(exc_info.value)
|
|
assert "insert()" in message
|
|
assert "await ainsert(" in message
|
|
# The previous fix (PR #3245) mislabelled this as a deadlock; it is an
|
|
# immediate RuntimeError. Keep the message free of that wording.
|
|
assert "deadlock" not in message.lower()
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_run_sync_does_not_create_coroutine_when_it_raises():
|
|
"""The factory is invoked lazily, so a guard failure never leaves an
|
|
un-awaited coroutine behind (which would emit a RuntimeWarning)."""
|
|
|
|
calls: list[bool] = []
|
|
|
|
def factory():
|
|
calls.append(True)
|
|
|
|
async def _coro():
|
|
return None
|
|
|
|
return _coro()
|
|
|
|
async def _inside_loop():
|
|
return _run_sync(factory, sync_name="query", async_name="aquery")
|
|
|
|
with pytest.raises(RuntimeError):
|
|
asyncio.run(_inside_loop())
|
|
|
|
assert calls == [], "coro_factory should not run when the guard rejects"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_run_sync_runs_when_owning_loop_matches():
|
|
"""When the loop being driven is the instance's owning loop, the coroutine
|
|
runs to completion — the cross-loop guard only fires on a mismatch."""
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
def factory():
|
|
async def _coro():
|
|
return 42
|
|
|
|
return _coro()
|
|
|
|
try:
|
|
# always_get_an_event_loop() returns the loop set above, which is the
|
|
# one we declare as owning -> match -> the coroutine runs.
|
|
result = _run_sync(
|
|
factory, sync_name="insert", async_name="ainsert", owning_loop=loop
|
|
)
|
|
assert result == 42
|
|
finally:
|
|
loop.close()
|
|
asyncio.set_event_loop(None)
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_run_sync_raises_when_driven_from_a_different_loop():
|
|
"""Driving a sync wrapper from a thread with no event loop of its own
|
|
(the ``loop.run_in_executor(None, rag.insert, …)`` anti-pattern) lands on a
|
|
freshly created loop, not the instance's owning loop. The guard must fail
|
|
fast pointing at the async alternative, without running the coroutine."""
|
|
|
|
owning_loop = asyncio.new_event_loop()
|
|
calls: list[bool] = []
|
|
|
|
def factory(): # pragma: no cover - must never be reached
|
|
calls.append(True)
|
|
raise AssertionError("coro_factory must not run on the wrong loop")
|
|
|
|
def call_off_thread():
|
|
# No loop is set on this worker thread, so always_get_an_event_loop()
|
|
# creates a brand-new one — different from owning_loop.
|
|
return _run_sync(
|
|
factory,
|
|
sync_name="insert",
|
|
async_name="ainsert",
|
|
owning_loop=owning_loop,
|
|
)
|
|
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=1) as executor:
|
|
with pytest.raises(RuntimeError) as exc_info:
|
|
executor.submit(call_off_thread).result()
|
|
finally:
|
|
owning_loop.close()
|
|
|
|
message = str(exc_info.value)
|
|
assert "insert()" in message
|
|
assert "await ainsert(" in message
|
|
assert "deadlock" not in message.lower()
|
|
assert calls == [], "coro_factory should not run on a mismatched loop"
|
|
|
|
|
|
@pytest.mark.offline
|
|
def test_run_sync_runs_when_owning_loop_is_closed():
|
|
"""A *closed* owning loop means the loop the storages were initialized on is
|
|
gone — the common ``rag = asyncio.run(initialize_rag())`` pattern, where
|
|
``asyncio.run`` closes the loop after ``initialize_storages()``. There is no
|
|
live loop to clash with, so the guard must let the call through and run the
|
|
coroutine on a fresh loop (preserving long-standing behavior), not reject
|
|
it."""
|
|
|
|
closed_owning_loop = asyncio.new_event_loop()
|
|
closed_owning_loop.close()
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
|
|
def factory():
|
|
async def _coro():
|
|
return 42
|
|
|
|
return _coro()
|
|
|
|
try:
|
|
result = _run_sync(
|
|
factory,
|
|
sync_name="insert",
|
|
async_name="ainsert",
|
|
owning_loop=closed_owning_loop,
|
|
)
|
|
assert result == 42
|
|
finally:
|
|
loop.close()
|
|
asyncio.set_event_loop(None)
|