Skip to content

Share downstream futures across timed window emitters - #504

Open
Aschenbath wants to merge 1 commit into
python-streamz:masterfrom
Aschenbath:fix/timed-window-shared-awaitable-284
Open

Aschenbath wants to merge 1 commit into
python-streamz:masterfrom
Aschenbath:fix/timed-window-shared-awaitable-284

Conversation

@Aschenbath

Copy link
Copy Markdown

When a timed window feeds an async def sink, the timer callback and subsequent inputs can await the same coroutine more than once, raising RuntimeError: cannot reuse already awaited coroutine. Convert each window's downstream results to a shared Future once, so concurrent and later inputs wait for the same work. Apply the same fix to timed_window_unique.

Fixes #284.

Nine regression/control cases cover both window types, both unique retention modes, native coroutines, Tasks, Tornado Futures, concurrent producers, completed work, two successive windows and a synchronous sink branch. On the original source, the three native-coroutine cases fail; all nine pass with the change on Windows CPython 3.10.22 and 3.12.15. Independent agent review passed 240 assertions, including exception propagation and synchronous callers using a background event loop. Flake8, wheel/sdist builds and installed-wheel checks pass.

The full suite is not green on this Windows host. The complete core module reports 130 passed / 4 failed with the patch; unchanged source reports 120 passed / 5 failed. All four patched failure cases also fail on the unchanged source across the baseline runs, involving existing timing/order assertions. Full-project runs on both versions reach 99% before the same test_process_str stalls and the 30-second watchdog stops execution. Other platforms and the upstream CI matrix were not run locally.

AI disclosure: prepared and tested with OpenAI Codex under the account owner's authorization, with independent review by a separate coding agent. No human review is claimed.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

'cannot reuse already awaited coroutine' with timed_window on Python 3.7.5

1 participant