From 4bc73ad0a381445dcb63bf318182082a471278e2 Mon Sep 17 00:00:00 2001 From: Heitor Neiva Date: Thu, 30 Jul 2026 16:26:54 -0700 Subject: [PATCH] fix: Gracefully handle event loops in fresh process In Python 3.12/3.13 they deprecated get_event_loop() returning a new loop when none is current. In Python 3.14 it now raises a RuntimeError. Reference: https://docs.python.org/3.14/library/asyncio-eventloop.html#asyncio.get_event_loop --- src/scriptworker/context.py | 9 ++- src/scriptworker/worker.py | 41 ++++++++---- tests/__init__.py | 24 +++++++ tests/conftest.py | 5 +- tests/test_context.py | 38 ++++++++++- tests/test_worker.py | 127 +++++++++++++++++++++++++++++++++++- 6 files changed, 225 insertions(+), 19 deletions(-) diff --git a/src/scriptworker/context.py b/src/scriptworker/context.py index 8a5d04df..856b9637 100644 --- a/src/scriptworker/context.py +++ b/src/scriptworker/context.py @@ -237,11 +237,18 @@ def projects(self, projects: Optional[Dict[str, Any]]) -> None: def event_loop(self) -> asyncio.AbstractEventLoop: """asyncio.BaseEventLoop: the running event loop. + Falls back to a newly created loop if none is running, since + ``asyncio.get_event_loop()`` raises rather than creating one when no + loop is current (python 3.14+). + This fixture mainly exists to allow for overrides during unit tests. """ if not self._event_loop: - self._event_loop = asyncio.get_event_loop() + try: + self._event_loop = asyncio.get_running_loop() + except RuntimeError: + self._event_loop = asyncio.new_event_loop() return self._event_loop @event_loop.setter diff --git a/src/scriptworker/worker.py b/src/scriptworker/worker.py index 4f370b71..6b455119 100644 --- a/src/scriptworker/worker.py +++ b/src/scriptworker/worker.py @@ -236,14 +236,23 @@ def main(event_loop=None): Args: event_loop (asyncio.BaseEventLoop, optional): the event loop to use. - If None, use ``asyncio.get_event_loop()``. Defaults to None. + If None, create one with ``asyncio.new_event_loop()`` and close it + on the way out. Defaults to None. """ context, credentials = get_context_from_cmdln(sys.argv[1:]) log.info("Scriptworker starting up at {} UTC".format(arrow.utcnow().format())) log.info("Worker FQDN: {}".format(socket.getfqdn())) cleanup(context) - context.event_loop = event_loop or asyncio.get_event_loop() + # ``asyncio.get_event_loop()`` raises rather than creating a loop when none + # is current (python 3.14+), so create one explicitly. Only close a loop we + # created ourselves; a caller-supplied loop stays open for the caller to use. + if event_loop is None: + owned_event_loop = asyncio.new_event_loop() + context.event_loop = owned_event_loop + else: + owned_event_loop = None + context.event_loop = event_loop done = False @@ -262,15 +271,21 @@ async def _handle_sigusr1(): log.info("SIGUSR1 received; no more tasks will be taken") done = True - context.event_loop.add_signal_handler(signal.SIGTERM, lambda: asyncio.ensure_future(_handle_sigterm())) - context.event_loop.add_signal_handler(signal.SIGUSR1, lambda: asyncio.ensure_future(_handle_sigusr1())) + try: + context.event_loop.add_signal_handler(signal.SIGTERM, lambda: asyncio.ensure_future(_handle_sigterm())) + context.event_loop.add_signal_handler(signal.SIGUSR1, lambda: asyncio.ensure_future(_handle_sigusr1())) - while not done: - try: - context.event_loop.run_until_complete(async_main(context, credentials)) - except Exception: - log.critical("Fatal exception", exc_info=1) - raise - else: - log.info("Scriptworker stopped at {} UTC".format(arrow.utcnow().format())) - log.info("Worker FQDN: {}".format(socket.getfqdn())) + while not done: + try: + context.event_loop.run_until_complete(async_main(context, credentials)) + except Exception: + log.critical("Fatal exception", exc_info=1) + raise + else: + log.info("Scriptworker stopped at {} UTC".format(arrow.utcnow().format())) + log.info("Worker FQDN: {}".format(socket.getfqdn())) + finally: + if owned_event_loop is not None: + # Closing the loop also deregisters the signal handlers added above, + # so repeated ``main()`` calls in one process don't pile them up. + owned_event_loop.close() diff --git a/tests/__init__.py b/tests/__init__.py index c6153728..318f560d 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -6,6 +6,7 @@ import json import os import sys +from contextlib import contextmanager import aiohttp import arrow @@ -46,6 +47,29 @@ def touch(path): print(path, file=fh, end="") +@contextmanager +def no_ambient_event_loop(): + """Run a block with no current event loop set for this thread. + + ``pytest-asyncio``'s auto mode installs a current event loop in the main + thread, so ``asyncio.get_event_loop()`` succeeds under pytest even where it + would raise in a clean process on python 3.14+. Tests that need to exercise + the no-current-loop path have to remove that ambient loop first. + + The previous loop is restored on the way out so sibling tests -- which run + in a random order -- still find the loop they expect. + """ + try: + previous_loop = asyncio.get_event_loop() + except RuntimeError: + previous_loop = None + asyncio.set_event_loop(None) + try: + yield + finally: + asyncio.set_event_loop(previous_loop) + + class FakeResponse(aiohttp.client_reqrep.ClientResponse): """Integration tests allow us to test everything's hooked up to aiohttp correctly. When we don't want to actually hit an external url, have diff --git a/tests/conftest.py b/tests/conftest.py index d14b907d..fa6b95cb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -171,5 +171,8 @@ def _craft_rw_context(tmp, cot_product, session, private=False): for rule in context.config["trusted_vcs_rules"]: rule["require_secret"] = True context.config["verbose"] = VERBOSE - context.event_loop = asyncio.new_event_loop() + # only reached from the async ``*_rw_context`` fixtures, so a loop is always + # running. Reusing it matches what ``Context.event_loop`` does in production + # and avoids leaking one unclosed loop per test. + context.event_loop = asyncio.get_running_loop() return context diff --git a/tests/test_context.py b/tests/test_context.py index eaad1f4f..91734b23 100644 --- a/tests/test_context.py +++ b/tests/test_context.py @@ -13,6 +13,8 @@ import scriptworker.context as swcontext from scriptworker.exceptions import CoTError +from . import no_ambient_event_loop + # constants helpers and fixtures {{{1 @pytest.fixture(scope="function") @@ -156,11 +158,41 @@ def test_get_credentials(rw_context): def test_new_event_loop(mocker): - """The default rw_context.event_loop is from `asyncio.get_event_loop`""" + """The default rw_context.event_loop is from `asyncio.new_event_loop`""" fake_loop = mock.MagicMock() - mocker.patch.object(asyncio, "get_event_loop", return_value=fake_loop) + mocker.patch.object(asyncio, "new_event_loop", return_value=fake_loop) rw_context = swcontext.Context() - assert rw_context.event_loop is fake_loop + with no_ambient_event_loop(): + assert rw_context.event_loop is fake_loop + + +def test_event_loop_no_ambient_loop(): + """`context.event_loop` creates a loop when none is current. + + Regression test for python 3.14, where `asyncio.get_event_loop()` raises + instead of creating a loop. + """ + rw_context = swcontext.Context() + with no_ambient_event_loop(): + loop = rw_context.event_loop + try: + assert isinstance(loop, asyncio.AbstractEventLoop) + # cached, so we don't create a fresh loop on every access + assert rw_context.event_loop is loop + finally: + loop.close() + + +@pytest.mark.asyncio +async def test_event_loop_prefers_running_loop(): + """`context.event_loop` returns the running loop rather than a new one. + + `create_queue` binds an aiohttp session to `context.event_loop`, so handing + back an idle loop while another one is running would bind the session to a + loop that never runs. + """ + rw_context = swcontext.Context() + assert rw_context.event_loop is asyncio.get_running_loop() def test_set_event_loop(mocker): diff --git a/tests/test_worker.py b/tests/test_worker.py index 3fefbdfb..45de9dc1 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -20,7 +20,18 @@ from scriptworker.exceptions import ScriptWorkerException, WorkerShutdownDuringTask from scriptworker.worker import RunTasks, do_run_task -from . import AT_LEAST_PY38, KILLED_SCRIPT, TIMEOUT_SCRIPT, create_async, create_finished_future, create_slow_async, create_sync, noop_async, noop_sync +from . import ( + AT_LEAST_PY38, + KILLED_SCRIPT, + TIMEOUT_SCRIPT, + create_async, + create_finished_future, + create_slow_async, + create_sync, + no_ambient_event_loop, + noop_async, + noop_sync, +) # constants helpers and fixtures {{{1 @@ -62,6 +73,120 @@ async def foo(arg, credentials): os.remove(tmp) +def test_main_no_ambient_event_loop(mocker, context): + """``main()`` creates its own loop when none is current, and closes it. + + Regression test for the python 3.14 startup crash: ``get_event_loop()`` no + longer falls back to creating a loop, so ``main()`` has to create one. + """ + config = dict(context.config) + config["poll_interval"] = 1 + config["credentials"] = {"fake_creds": True} + + loops = [] + + async def foo(arg, credentials): + loops.append(arg.event_loop) + raise ScriptWorkerException("foo") + + _, tmp = tempfile.mkstemp() + try: + with open(tmp, "w") as fh: + json.dump(config, fh) + del config["credentials"] + mocker.patch.object(worker, "async_main", new=foo) + mocker.patch.object(sys, "argv", new=["x", tmp]) + with no_ambient_event_loop(): + with pytest.raises(ScriptWorkerException): + worker.main() + finally: + os.remove(tmp) + + assert len(loops) == 1 + # the loop main() created is its own, and it cleaned up after itself + assert loops[0].is_closed() + + +def test_main_does_not_close_caller_loop(mocker, context): + """``main()`` leaves a caller-supplied loop open for the caller to keep using.""" + event_loop = asyncio.new_event_loop() + + async def foo(arg, credentials): + assert arg.event_loop is event_loop + raise ScriptWorkerException("foo") + + config = dict(context.config) + config["poll_interval"] = 1 + config["credentials"] = {"fake_creds": True} + + _, tmp = tempfile.mkstemp() + try: + with open(tmp, "w") as fh: + json.dump(config, fh) + mocker.patch.object(worker, "async_main", new=foo) + mocker.patch.object(sys, "argv", new=["x", tmp]) + with pytest.raises(ScriptWorkerException): + worker.main(event_loop=event_loop) + assert not event_loop.is_closed() + finally: + event_loop.close() + os.remove(tmp) + + +def test_main_repeated_calls_do_not_leak_loops(mocker, context): + """Restarting ``main()`` in one process doesn't accumulate open loops.""" + config = dict(context.config) + config["poll_interval"] = 1 + config["credentials"] = {"fake_creds": True} + + loops = [] + + async def foo(arg, credentials): + loops.append(arg.event_loop) + raise ScriptWorkerException("foo") + + _, tmp = tempfile.mkstemp() + try: + with open(tmp, "w") as fh: + json.dump(config, fh) + mocker.patch.object(worker, "async_main", new=foo) + mocker.patch.object(sys, "argv", new=["x", tmp]) + with no_ambient_event_loop(): + for _ in range(3): + with pytest.raises(ScriptWorkerException): + worker.main() + finally: + os.remove(tmp) + + assert len(loops) == 3 + assert all(loop.is_closed() for loop in loops) + + +def test_main_closes_loop_when_signal_handler_setup_fails(mocker, context): + """A failure while arming the signal handlers still closes the loop we made.""" + created = asyncio.new_event_loop() + mocker.patch.object(asyncio, "new_event_loop", return_value=created) + mocker.patch.object(created, "add_signal_handler", side_effect=NotImplementedError("no signals here")) + + config = dict(context.config) + config["poll_interval"] = 1 + config["credentials"] = {"fake_creds": True} + + _, tmp = tempfile.mkstemp() + try: + with open(tmp, "w") as fh: + json.dump(config, fh) + mocker.patch.object(worker, "async_main", new=noop_async) + mocker.patch.object(sys, "argv", new=["x", tmp]) + with no_ambient_event_loop(): + with pytest.raises(NotImplementedError): + worker.main() + finally: + os.remove(tmp) + + assert created.is_closed() + + @pytest.mark.parametrize("running", (True, False)) def test_main_running_sigterm(mocker, context, running): """Test that sending SIGTERM causes the main loop to stop after the next