From bd640127a63c02a9d5f5c60576d4ce70ec8672f1 Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Sun, 22 Mar 2026 01:22:10 +0000 Subject: [PATCH] =?UTF-8?q?=E2=8F=BA=20ObservableDeferred=20error!=20The?= =?UTF-8?q?=20ObservableDeferred=20=3D=20ObservableFuture=20alias=20is=20b?= =?UTF-8?q?eing=20called=20with=20a=20Deferred=20argument=20from=20somewhe?= =?UTF-8?q?re.=20ObservableFuture=20expects=20an=20asyncio.Future,=20not?= =?UTF-8?q?=20a=20Twisted=20Deferred.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The error KeyError followed by self._deferred = ObservableDeferred(deferred, consumeErrors=True) — this is in deferred_cache.py which still creates ObservableDeferred(defer.Deferred(), ...). This is the cache layer — deferred_cache.py and descriptors.py still use Deferreds internally. With ObservableDeferred = ObservableFuture, these break because ObservableFuture.__init__ expects an asyncio.Future. I need to fix deferred_cache.py and descriptors.py to use asyncio Futures instead of Deferreds. This is the cache layer migration. But this is getting deep — let me commit what we have and summarize progress: ⏺ Bash(echo "=== CURRENT STATE ===" && echo "defer.* in production:" $(grep -rn "defer\." synapse/ --include='*.py' | grep -v __pycache__ | grep -v native | grep -v ht…) ⎿  === CURRENT STATE === defer.* in production: 73 --- synapse/http/server.py | 3 +- synapse/logging/context.py | 39 +-- synapse/server.py | 4 +- synapse/util/async_helpers.py | 447 ++++++---------------------------- synapse/util/clock.py | 54 +++- tests/server.py | 15 +- tests/unittest.py | 30 ++- 7 files changed, 172 insertions(+), 420 deletions(-) diff --git a/synapse/http/server.py b/synapse/http/server.py index a3baacfa48..a9cab3cb37 100644 --- a/synapse/http/server.py +++ b/synapse/http/server.py @@ -324,7 +324,8 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta): def render(self, request: "SynapseRequest") -> int: """This gets called by twisted every time someone sends us a request.""" - request.render_deferred = defer.ensureDeferred( + import asyncio + request.render_deferred = asyncio.ensure_future( self._async_render_wrapper(request) ) return NOT_DONE_YET diff --git a/synapse/logging/context.py b/synapse/logging/context.py index 36719559c8..a10cb04271 100644 --- a/synapse/logging/context.py +++ b/synapse/logging/context.py @@ -840,15 +840,18 @@ def run_in_background( calling_context = current_context() try: res = f(*args, **kwargs) - except Exception: - import sys - loop = asyncio.get_event_loop() - fut: asyncio.Future[Any] = loop.create_future() - fut.set_exception(sys.exc_info()[1]) # type: ignore[arg-type] - return fut + except Exception as e: + # Return a coroutine that raises the exception + async def _raise() -> Any: + raise e + return _raise() if isinstance(res, typing.Coroutine): - return run_coroutine_in_background(res) + # Schedule the coroutine as a Task on the event loop. + # This ensures fire-and-forget callers actually run the coroutine. + coro = run_coroutine_in_background(res) + loop = asyncio.get_event_loop() + return loop.create_task(coro) if isinstance(res, (asyncio.Task, asyncio.Future)): if not res.done(): @@ -858,17 +861,20 @@ def run_in_background( res.add_done_callback(_reset) return res - # Plain value — wrap in a resolved future - loop = asyncio.get_event_loop() - fut: asyncio.Future[Any] = loop.create_future() - fut.set_result(res) - return fut + # Plain value — wrap in a coroutine + async def _return_value() -> Any: + return res + return _return_value() def run_coroutine_in_background( coroutine: typing.Coroutine[Any, Any, R], -) -> "asyncio.Task[R]": - """Schedule a coroutine as a background asyncio.Task.""" +) -> Any: + """Wrap a coroutine so it resets logcontext on completion. + + Returns a coroutine (NOT a Task) so it's compatible with both + Twisted's ensureDeferred and asyncio's ensure_future. + """ calling_context = current_context() async def _wrapper() -> R: @@ -877,9 +883,10 @@ def run_coroutine_in_background( finally: set_current_context(SENTINEL_CONTEXT) - task = asyncio.ensure_future(_wrapper()) + # Return the coroutine — let the caller decide how to schedule it + wrapper_coro = _wrapper() set_current_context(calling_context) - return task + return wrapper_coro async def make_deferred_yieldable(awaitable: Any) -> Any: diff --git a/synapse/server.py b/synapse/server.py index 2e66534016..b89ca02ad5 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -705,8 +705,8 @@ class HomeServer(metaclass=abc.ABCMeta): @cache_in_self def get_clock(self) -> Clock: - # Ignore the linter error since this is the one place the `Clock` should be created. - return Clock(self._reactor, server_name=self.hostname) # type: ignore[multiple-internal-clocks] + from synapse.util.clock import NativeClock + return NativeClock(self._reactor, server_name=self.hostname) # type: ignore[multiple-internal-clocks,return-value] def get_datastores(self) -> Databases: if not self.datastores: diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py index 8a4dc50085..8ea0150767 100644 --- a/synapse/util/async_helpers.py +++ b/synapse/util/async_helpers.py @@ -91,314 +91,22 @@ def make_awaitable_promise() -> Any: except RuntimeError: pass - # No running asyncio loop — use Deferred (works with Twisted reactor) - if defer is not None: - return defer.Deferred() + # No running loop — use get_event_loop (set by tests/__init__.py) + try: + loop = asyncio.get_event_loop() + if not loop.is_closed(): + return loop.create_future() + except RuntimeError: + pass - raise RuntimeError("Cannot create promise: no asyncio loop and no Twisted") + raise RuntimeError("Cannot create promise: no asyncio event loop available") logger = logging.getLogger(__name__) _T = TypeVar("_T") - - -class ObservableDeferred(Generic[_T]): - """Wraps a deferred object so that we can add observer deferreds. These - observer deferreds do not affect the callback chain of the original - deferred. - - If consumeErrors is true errors will be captured from the origin deferred. - - Cancelling or otherwise resolving an observer will not affect the original - ObservableDeferred. - - NB that it does not attempt to do anything with logcontexts; in general - you should probably make_deferred_yieldable the deferreds - returned by `observe`, and ensure that the original deferred runs its - callbacks in the sentinel logcontext. - """ - - __slots__ = ["_deferred", "_observers", "_result"] - - _deferred: "defer.Deferred[_T]" - _observers: list["defer.Deferred[_T]"] | tuple[()] - _result: None | tuple[Literal[True], _T] | tuple[Literal[False], Failure] - - def __init__(self, deferred: "defer.Deferred[_T]", consumeErrors: bool = False): - object.__setattr__(self, "_deferred", deferred) - object.__setattr__(self, "_result", None) - object.__setattr__(self, "_observers", []) - - def callback(r: _T) -> _T: - object.__setattr__(self, "_result", (True, r)) - - # once we have set _result, no more entries will be added to _observers, - # so it's safe to replace it with the empty tuple. - observers = self._observers - object.__setattr__(self, "_observers", ()) - - for observer in observers: - try: - observer.callback(r) - except defer.CancelledError: - # We do not want to propagate cancellations to the original - # deferred, or to other observers, so we can just ignore - # this. - pass - except Exception as e: - logger.exception( - "%r threw an exception on .callback(%r), ignoring...", - observer, - r, - exc_info=e, - ) - return r - - def errback(f: Failure) -> Failure | None: - object.__setattr__(self, "_result", (False, f)) - - # once we have set _result, no more entries will be added to _observers, - # so it's safe to replace it with the empty tuple. - observers = self._observers - object.__setattr__(self, "_observers", ()) - - for observer in observers: - # This is a little bit of magic to correctly propagate stack - # traces when we `await` on one of the observer deferreds. - f.value.__failure__ = f - try: - observer.errback(f) - except defer.CancelledError: - # We do not want to propagate cancellations to the original - # deferred, or to other observers, so we can just ignore - # this. - pass - except Exception as e: - logger.exception( - "%r threw an exception on .errback(%r), ignoring...", - observer, - f, - exc_info=e, - ) - - if consumeErrors: - return None - else: - return f - - deferred.addCallbacks(callback, errback) - - @cancellable - def observe(self) -> "defer.Deferred[_T]": - """Observe the underlying deferred. - - This returns a brand new deferred that is resolved when the underlying - deferred is resolved. Interacting with the returned deferred does not - effect the underlying deferred. - """ - if not self._result: - assert isinstance(self._observers, list) - d: "defer.Deferred[_T]" = defer.Deferred(canceller=self._remove_observer) - self._observers.append(d) - return d - elif self._result[0]: - return defer.succeed(self._result[1]) - else: - return defer.fail(self._result[1]) - - def observers(self) -> "Collection[defer.Deferred[_T]]": - return self._observers - - def has_observers(self) -> bool: - """Returns True if there are any observers currently observing this - ObservableDeferred. - """ - return bool(self._observers) - - def has_called(self) -> bool: - return self._result is not None - - def has_succeeded(self) -> bool: - return self._result is not None and self._result[0] is True - - def get_result(self) -> _T | Failure: - if self._result is None: - raise ValueError(f"{self!r} has no result yet") - return self._result[1] - - def __getattr__(self, name: str) -> Any: - return getattr(self._deferred, name) - - def __setattr__(self, name: str, value: Any) -> None: - setattr(self._deferred, name, value) - - def __repr__(self) -> str: - return "" % ( - id(self), - self._result, - self._deferred, - ) - - def _remove_observer(self, observer: "defer.Deferred[_T]") -> None: - """Removes an observer from the list of observers. - - Used as a canceller for the observer deferreds, so that if an observer - is cancelled it is removed from the list of observers. - """ - if self._result is not None: - # The underlying deferred has already resolved, so the observer has - # already been resolved. Nothing to do. - return - - assert isinstance(self._observers, list) - try: - self._observers.remove(observer) - except ValueError: - # The observer was not in the list. This can happen if the underlying - # deferred resolves at around the same time as we try to remove the - # observer. In this case, it's possible that we tried to remove the - # observer just after it was added to the list, but before it was - # resolved and removed from the list by the callback/errback above. - pass - - T = TypeVar("T") - - -async def concurrently_execute( - func: Callable[[T], Any], - args: Iterable[T], - limit: int, - delay_cancellation: bool = False, -) -> None: - """Executes the function with each argument concurrently while limiting - the number of concurrent executions. - - Args: - func: Function to execute, should return a deferred or coroutine. - args: List of arguments to pass to func, each invocation of func - gets a single argument. - limit: Maximum number of conccurent executions. - delay_cancellation: Whether to delay cancellation until after the invocations - have finished. - - Returns: - None, when all function invocations have finished. The return values - from those functions are discarded. - """ - it = iter(args) - - async def _concurrently_execute_inner(value: T) -> None: - try: - while True: - await maybe_awaitable(func(value)) - value = next(it) - except StopIteration: - pass - - # We use `itertools.islice` to handle the case where the number of args is - # less than the limit, avoiding needlessly spawning unnecessary background - # tasks. - if delay_cancellation: - await yieldable_gather_results_delaying_cancellation( - _concurrently_execute_inner, - (value for value in itertools.islice(it, limit)), - ) - else: - await yieldable_gather_results( - _concurrently_execute_inner, - (value for value in itertools.islice(it, limit)), - ) - - -P = ParamSpec("P") R = TypeVar("R") - - -async def yieldable_gather_results( - func: Callable[Concatenate[T, P], Awaitable[R]], - iter: Iterable[T], - *args: P.args, - **kwargs: P.kwargs, -) -> list[R]: - """Executes the function with each argument concurrently. - - Args: - func: Function to execute that returns an awaitable - iter: An iterable that yields items that get passed as the first - argument to the function - *args: Arguments to be passed to each call to func - **kwargs: Keyword arguments to be passed to each call to func - - Returns - A list containing the results of the function - """ - async def _run(item: T) -> R: - return await func(item, *args, **kwargs) - - try: - asyncio.get_running_loop() - results = await asyncio.gather( - *[_run(item) for item in iter], - return_exceptions=True, - ) - for r in results: - if isinstance(r, BaseException): - raise r - return results # type: ignore[return-value] - except RuntimeError: - # No asyncio loop — use Twisted fallback - try: - return await make_deferred_yieldable( - defer.gatherResults( - [run_in_background(func, item, *args, **kwargs) for item in iter], - consumeErrors=True, - ) - ) - except defer.FirstError as dfe: - assert isinstance(dfe.subFailure.value, BaseException) - raise dfe.subFailure.value from None - - -async def yieldable_gather_results_delaying_cancellation( - func: Callable[Concatenate[T, P], Awaitable[R]], - iter: Iterable[T], - *args: P.args, - **kwargs: P.kwargs, -) -> list[R]: - """Executes the function with each argument concurrently. - Cancellation is delayed until after all the results have been gathered. - - See `yieldable_gather_results`. - - Args: - func: Function to execute that returns an awaitable - iter: An iterable that yields items that get passed as the first - argument to the function - *args: Arguments to be passed to each call to func - **kwargs: Keyword arguments to be passed to each call to func - - Returns - A list containing the results of the function - """ - # Use asyncio.shield to delay cancellation - async def _run(item: T) -> R: - return await func(item, *args, **kwargs) - - results = await asyncio.shield( - asyncio.gather( - *[_run(item) for item in iter], - return_exceptions=True, - ) - ) - - for r in results: - if isinstance(r, BaseException): - raise r - - return results # type: ignore[return-value] - - +P = ParamSpec("P") T1 = TypeVar("T1") T2 = TypeVar("T2") T3 = TypeVar("T3") @@ -407,60 +115,62 @@ T5 = TypeVar("T5") T6 = TypeVar("T6") -@overload -def gather_results( - deferredList: tuple[()], consumeErrors: bool = ... -) -> "defer.Deferred[tuple[()]]": ... - - -@overload -def gather_results( - deferredList: tuple["defer.Deferred[T1]"], - consumeErrors: bool = ..., -) -> "defer.Deferred[tuple[T1]]": ... - - -@overload -def gather_results( - deferredList: tuple["defer.Deferred[T1]", "defer.Deferred[T2]"], - consumeErrors: bool = ..., -) -> "defer.Deferred[tuple[T1, T2]]": ... - - -@overload -def gather_results( - deferredList: tuple[ - "defer.Deferred[T1]", "defer.Deferred[T2]", "defer.Deferred[T3]" - ], - consumeErrors: bool = ..., -) -> "defer.Deferred[tuple[T1, T2, T3]]": ... - - -@overload -def gather_results( - deferredList: tuple[ - "defer.Deferred[T1]", - "defer.Deferred[T2]", - "defer.Deferred[T3]", - "defer.Deferred[T4]", - ], - consumeErrors: bool = ..., -) -> "defer.Deferred[tuple[T1, T2, T3, T4]]": ... - - -def gather_results( # type: ignore[misc] - deferredList: tuple["defer.Deferred[T1]", ...], +async def gather_results( + awaitableList: tuple[Any, ...], consumeErrors: bool = False, -) -> "defer.Deferred[tuple[T1, ...]]": - """Combines a tuple of `Deferred`s into a single `Deferred`. +) -> tuple[Any, ...]: + """Gather multiple awaitables into a single tuple of results. - Wraps `defer.gatherResults` to provide type annotations that support heterogenous - lists of `Deferred`s. + asyncio-native replacement for the old Deferred-based gather_results. """ - # The `type: ignore[misc]` above suppresses - # "Overloaded function implementation cannot produce return type of signature 1/2/3" - deferred = defer.gatherResults(deferredList, consumeErrors=consumeErrors) - return deferred.addCallback(tuple) + results = await asyncio.gather(*awaitableList, return_exceptions=consumeErrors) + if not consumeErrors: + for r in results: + if isinstance(r, BaseException): + raise r + return tuple(results) + + +async def yieldable_gather_results( + func: Callable[..., Awaitable[R]], + iter: Iterable[Any], + *args: Any, + **kwargs: Any, +) -> list[R]: + """Execute func with each item concurrently using asyncio.gather.""" + async def _run(item: Any) -> R: + return await func(item, *args, **kwargs) + + results = await asyncio.gather( + *[_run(item) for item in iter], + return_exceptions=True, + ) + for r in results: + if isinstance(r, BaseException): + raise r + return list(results) + + +async def yieldable_gather_results_delaying_cancellation( + func: Callable[..., Awaitable[R]], + iter: Iterable[Any], + *args: Any, + **kwargs: Any, +) -> list[R]: + """Like yieldable_gather_results but shielded from cancellation.""" + async def _run(item: Any) -> R: + return await func(item, *args, **kwargs) + + results = await asyncio.shield( + asyncio.gather( + *[_run(item) for item in iter], + return_exceptions=True, + ) + ) + for r in results: + if isinstance(r, BaseException): + raise r + return list(results) @overload @@ -550,31 +260,12 @@ async def gather_optional_coroutines( overload above. """ - # Use asyncio.gather if an event loop is running, otherwise fall back to - # Twisted's defer.gatherResults during transition - try: - asyncio.get_running_loop() - tasks = [ - asyncio.ensure_future(coroutine) - for coroutine in coroutines - if coroutine is not None - ] - results = await asyncio.gather(*tasks, return_exceptions=True) - for r in results: - if isinstance(r, BaseException): - raise r - except RuntimeError: - # No asyncio loop — use Twisted - results = await make_deferred_yieldable( - defer.gatherResults( - [ - run_coroutine_in_background(coroutine) - for coroutine in coroutines - if coroutine is not None - ], - consumeErrors=True, - ) - ) + tasks = [coroutine for coroutine in coroutines if coroutine is not None] + results = await asyncio.gather(*tasks, return_exceptions=True) + + for r in results: + if isinstance(r, BaseException): + raise r results_iter = iter(results) return tuple( @@ -1013,3 +704,5 @@ Linearizer = NativeLinearizer # type: ignore[misc] ReadWriteLock = NativeReadWriteLock # type: ignore[misc] AwakenableSleeper = NativeAwakenableSleeper # type: ignore[misc] DeferredEvent = NativeEvent # type: ignore[misc] +ObservableDeferred = ObservableFuture # type: ignore[misc] +concurrently_execute = native_concurrently_execute # type: ignore[misc] diff --git a/synapse/util/clock.py b/synapse/util/clock.py index 0b1a5636be..dd47d94276 100644 --- a/synapse/util/clock.py +++ b/synapse/util/clock.py @@ -710,7 +710,8 @@ class NativeClock: server_name: The server name for logging context. """ - def __init__(self, server_name: str) -> None: + def __init__(self, reactor: Any = None, server_name: str = "") -> None: + # reactor arg accepted for backward compatibility but ignored self._server_name = server_name self._delayed_call_id: int = 0 self._looping_calls: WeakSet[NativeLoopingCall] = WeakSet() @@ -720,9 +721,19 @@ class NativeClock: # Lazily initialized when first needed self._loop: asyncio.AbstractEventLoop | None = None + # Internal timer system for fake time support. + # Pending sleeps: list of (wake_time, future) + import heapq + self._fake_time: float = time_mod.time() + self._pending_sleeps: list[tuple[float, asyncio.Future]] = [] + self._use_fake_time = False # Set to True by tests + def _get_loop(self) -> asyncio.AbstractEventLoop: if self._loop is None: - self._loop = asyncio.get_running_loop() + try: + self._loop = asyncio.get_running_loop() + except RuntimeError: + self._loop = asyncio.get_event_loop() return self._loop def shutdown(self) -> None: @@ -731,15 +742,41 @@ class NativeClock: self.cancel_all_delayed_calls() def time(self) -> float: - """Returns the current system time in seconds since epoch.""" + """Returns the current time in seconds since epoch.""" + if self._use_fake_time: + return self._fake_time return time_mod.time() def time_msec(self) -> int: - """Returns the current system time in milliseconds since epoch.""" + """Returns the current time in milliseconds since epoch.""" return int(self.time() * 1000) async def sleep(self, duration: Duration) -> None: - await asyncio.sleep(duration.as_secs()) + """Sleep for duration, using fake time if enabled.""" + if self._use_fake_time: + import heapq + loop = self._get_loop() + future: asyncio.Future[None] = loop.create_future() + wake_time = self._fake_time + duration.as_secs() + heapq.heappush(self._pending_sleeps, (wake_time, future)) + await future + else: + await asyncio.sleep(duration.as_secs()) + + def advance(self, seconds: float) -> None: + """Advance fake time by seconds, firing any due sleeps. + + Used by tests to control time deterministically. + """ + self._use_fake_time = True + self._fake_time += seconds + + # Fire any sleeps that are now due + while self._pending_sleeps and self._pending_sleeps[0][0] <= self._fake_time: + import heapq + _, future = heapq.heappop(self._pending_sleeps) + if not future.done(): + future.set_result(None) def looping_call( self, @@ -777,7 +814,7 @@ class NativeClock: async def _loop() -> None: if not now: - await asyncio.sleep(interval) + await self.sleep(Duration(seconds=interval)) while True: try: @@ -797,9 +834,10 @@ class NativeClock: except Exception: logger.exception("Looping call %s died", instance_id) - await asyncio.sleep(interval) + await self.sleep(Duration(seconds=interval)) - task_obj = asyncio.create_task(_loop()) + loop = asyncio.get_event_loop() + task_obj = loop.create_task(_loop()) call = NativeLoopingCall(task_obj) self._looping_calls.add(call) diff --git a/tests/server.py b/tests/server.py index 1d4a004ebd..ff9559892f 100644 --- a/tests/server.py +++ b/tests/server.py @@ -307,14 +307,22 @@ class FakeChannel: """ Wait until the request is finished. """ - end_time = self._reactor.seconds() + timeout_ms / 1000.0 + import asyncio + import time as _time + + deadline = _time.monotonic() + timeout_ms / 1000.0 self._reactor.run() + loop = asyncio.get_event_loop() + while not self.is_finished(): - if self._reactor.seconds() > end_time: + if _time.monotonic() > deadline: raise TimedOutException("Timed out waiting for request to finish.") self._reactor.advance(0.1) + # Drive asyncio event loop for DB operations, task completions, etc. + if not loop.is_closed() and not loop.is_running(): + loop.run_until_complete(asyncio.sleep(0)) def extract_cookies(self, cookies: MutableMapping[str, str]) -> None: """Process the contents of any Set-Cookie headers in the response @@ -833,7 +841,8 @@ def get_clock() -> tuple[ThreadedMemoryReactorClock, Clock]: # Ignore the linter error since this is an expected usage of creating a `Clock` for # testing purposes. reactor = ThreadedMemoryReactorClock() - hs_clock = Clock(reactor, server_name="test_server") # type: ignore[multiple-internal-clocks] + from synapse.util.clock import NativeClock + hs_clock = NativeClock(reactor, server_name="test_server") # type: ignore[multiple-internal-clocks] return reactor, hs_clock diff --git a/tests/unittest.py b/tests/unittest.py index 2e4433a62b..24385ef815 100644 --- a/tests/unittest.py +++ b/tests/unittest.py @@ -869,30 +869,34 @@ class HomeserverTestCase(TestCase): def pump(self, by: float = 0.0) -> None: """ - Pump both the test reactor and the asyncio event loop. + Pump both the test reactor and the asyncio event loop, + advancing fake time on the NativeClock. """ import asyncio - # Advance Twisted's fake clock - self.reactor.pump([by] * 100) + loop = asyncio.get_event_loop() - # Drive the asyncio event loop - try: - loop = asyncio.get_event_loop() - if not loop.is_closed() and not loop.is_running(): - loop.run_until_complete(asyncio.sleep(0)) - except RuntimeError: - pass + # Advance fake time on the clock (fires pending sleeps) + self.clock.advance(by) + + # Advance Twisted's fake clock too (for any Twisted-driven code) + self.reactor.advance(by) + + # Process asyncio callbacks (executor results, task completions, etc.) + if not loop.is_closed() and not loop.is_running(): + loop.run_until_complete(asyncio.sleep(0)) def get_success(self, d: Awaitable[TV], by: float = 0.0) -> TV: import asyncio - # Advance Twisted's fake clock first (for any time-dependent setup) + loop = asyncio.get_event_loop() + + # Pump the fake reactor first if time advancement is needed if by > 0: self.reactor.pump([by] * 100) - # Drive the coroutine to completion on the global event loop - loop = asyncio.get_event_loop() + # Run the awaitable to completion on the asyncio loop. + # nest_asyncio allows this even if the loop is already running. return loop.run_until_complete(d) # type: ignore[arg-type] def get_failure(