fix: Gracefully handle event loops in fresh process

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

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()

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

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

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):

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