From d60dfa5ed8e7480fe8c1252f181725a9508efdc0 Mon Sep 17 00:00:00 2001 From: Beau Harrison Date: Wed, 12 Aug 2026 15:30:04 -0500 Subject: [PATCH 1/2] fix: run DPM reply timeout in task --- .gitignore | 1 - acsys/dpm/__init__.py | 5 ++- tests/test_dpm_task_context.py | 60 ++++++++++++++++++++++++++++++++++ 3 files changed, 64 insertions(+), 2 deletions(-) create mode 100644 tests/test_dpm_task_context.py diff --git a/.gitignore b/.gitignore index 1deb341..7e16da1 100644 --- a/.gitignore +++ b/.gitignore @@ -140,6 +140,5 @@ cython_debug/ # Application specific acsys.tgz -tests/ .python-version .vscode diff --git a/acsys/dpm/__init__.py b/acsys/dpm/__init__.py index e56ddfe..9b85801 100644 --- a/acsys/dpm/__init__.py +++ b/acsys/dpm/__init__.py @@ -411,7 +411,10 @@ async def replies(self, tmo=None): """ while True: - ii = await asyncio.wait_for(self.__anext__(), tmo) + reply_task = asyncio.create_task(self.__anext__()) + timeout_task = asyncio.create_task( + asyncio.wait_for(reply_task, tmo)) + ii = await timeout_task if ii is None: return yield ii diff --git a/tests/test_dpm_task_context.py b/tests/test_dpm_task_context.py new file mode 100644 index 0000000..d0431ce --- /dev/null +++ b/tests/test_dpm_task_context.py @@ -0,0 +1,60 @@ +import asyncio +import pytest +import acsys.dpm + + +def _new_loop(): + """Create a test loop without leaking the loop created during import.""" + try: + current_loop = asyncio.get_event_loop() + except RuntimeError: + current_loop = None + if current_loop is not None and not current_loop.is_running(): + current_loop.close() + + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + return loop + + +def _drive_once_without_task(coro, loop): + """Advance a coroutine once without wrapping it in an asyncio.Task.""" + asyncio.events._set_running_loop(loop) + try: + return coro.send(None) + finally: + asyncio.events._set_running_loop(None) + + +def _close_loop(loop): + pending = asyncio.all_tasks(loop) + for task in pending: + task.cancel() + if pending: + loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) + asyncio.set_event_loop(None) + loop.close() + + +def test_dpm_replies_timeout_uses_task_context(): + """DPM.replies() can start without an existing asyncio.Task.""" + dpm = object.__new__(acsys.dpm.DPM) + + async def dummy_anext(): + await asyncio.sleep(0.01) + return "data" + + dpm.__anext__ = dummy_anext + loop = _new_loop() + coro = dpm.replies(tmo=1.0).__anext__() + + try: + timeout_task = _drive_once_without_task(coro, loop) + assert isinstance(timeout_task, asyncio.Task) + assert loop.run_until_complete(timeout_task) == "data" + with pytest.raises(StopIteration) as stopped: + coro.send(None) + assert stopped.value.value == "data" + finally: + coro.close() + _close_loop(loop) From 664ea92cfeca3a5f6624fd09efcd2b052dd94b37 Mon Sep 17 00:00:00 2001 From: Beau Harrison Date: Wed, 12 Aug 2026 15:30:15 -0500 Subject: [PATCH 2/2] docs: record DPM timeout fix status --- ...bugfix-dpm-replies-timeout-task-context.md | 61 +++++++++++++++++++ 1 file changed, 61 insertions(+) create mode 100644 .github/WORKFLOW_TASKS/issue-2-bugfix-dpm-replies-timeout-task-context.md diff --git a/.github/WORKFLOW_TASKS/issue-2-bugfix-dpm-replies-timeout-task-context.md b/.github/WORKFLOW_TASKS/issue-2-bugfix-dpm-replies-timeout-task-context.md new file mode 100644 index 0000000..3a386ae --- /dev/null +++ b/.github/WORKFLOW_TASKS/issue-2-bugfix-dpm-replies-timeout-task-context.md @@ -0,0 +1,61 @@ +# Issue 2: [Bugfix] Resolve `RuntimeError('Timeout should be used inside a task')` in `acsys.dpm.replies()` on Python 3.11+ + +**Status:** Implemented on branch `fix/dpm-replies-task-context`; full pytest suite passes locally. Draft PR pending. + +## Summary + +Make `acsys.dpm.DPM.replies()` usable when its async generator is driven through an event loop on Python 3.11 and later, without exposing the `asyncio.wait_for()` task-context runtime error. + +## Reproduction and current evidence + +`tests/test_dpm_task_context.py` reproduces the problem in the RED phase by: + +1. Creating a bare `DPM` instance with a stubbed `__anext__`. +2. Creating `replies_gen = dpm.replies(tmo=1.0)`. +3. Driving the async generator once with `.send(None)` while marking a bare event loop as running, without wrapping the generator in an active task. +4. Asserting that Python 3.11+ raises `RuntimeError` matching `Timeout should be used inside a task`. + +The GREEN regression test continues the manually driven generator after the fix and verifies that the first yielded object is a task and that the reply is returned successfully. + +The relevant implementation is `DPM.replies()` in `acsys/dpm/__init__.py`, where each reply is awaited through `asyncio.wait_for(self.__anext__(), tmo)`. + +## Scope + +- Diagnose the supported invocation patterns for `DPM.replies()` and the Python 3.11+ `asyncio` requirement involved in the failure. +- Change the implementation and/or surrounding async boundary so the public replies API no longer fails solely because the caller drives the async generator without an existing task. +- Preserve timeout semantics: `tmo` remains the maximum interval between replies and an `asyncio.TimeoutError` remains the expected timeout signal. +- Replace the current regression test's failure expectation with a success/behavioral regression test that covers the supported calling pattern. + +## Acceptance criteria + +- The reproducer no longer raises `RuntimeError('Timeout should be used inside a task')` on Python 3.11+. +- Iterating `DPM.replies(tmo=...)` from a normal task continues to yield replies as before. +- A timeout between replies still raises `asyncio.TimeoutError` with the documented behavior. +- Cancellation and generator cleanup do not leave pending tasks or unhandled coroutine warnings. +- The regression coverage runs under the repository's supported Python versions, or any version-specific limitation is explicitly documented. +- No unrelated changes to DPM request/reconnection behavior are introduced. + +## Verification + +- RED phase: the `.send(None)` reproducer failed before the implementation change with `RuntimeError('Timeout should be used inside a task')`. +- GREEN phase: `python -m pytest -q tests/test_dpm_task_context.py -W error::RuntimeWarning` passed. +- Full suite: `python -m pytest -q` passed (`1 passed`). +- Focused regression test passed with runtime warnings treated as errors. +- The fix creates a task for `self.__anext__()` and a task for the `asyncio.wait_for()` coroutine, keeping Python 3.11+ timeout context execution inside a task. + +## Dependencies and risks + +- Issue 1 should establish the pytest/pytest-asyncio configuration before this issue is implemented. +- The exact repair must preserve the caller's event-loop ownership and avoid silently creating a competing loop. +- The test's current bare-object setup bypasses normal `DPM` initialization; implementation should also be validated through the real public setup path where practical. + +## Review focus + +- Whether the fix addresses the task-context boundary rather than masking the exception. +- Whether timeout, cancellation, and async-generator lifecycle semantics remain correct. +- Whether tests assert observable API behavior instead of depending only on a particular `asyncio` implementation detail. + +## Out of scope + +- Replacing the synchronization primitives used by `set_many()` (Issue 3). +- General modernization of all `asyncio.get_event_loop()` calls in the package.