memory (disco): Fix pending queries isolation and cancellation:

When returning pending queries `Deferred`, a caller could add its own callbacks/errbacks
and modify the result. When another caller was reusing the `Deferred`, it got the modified
result (which could be `None`) instead of the expected `DiscoInfo`.

This has been fixed by creating and new `Deferred` on each request, and chaining it to the
original pending queries using the new `fork_deferred` helper method, so each caller get
its own isolated `Deferred`.

In `fork_deferred`, `cancel` method is used to call the parent Deferred `cancel`, to avoid
a wrong Exception propagation and the chained Deferred to be eventually called twice
(triggering `AlreadyCalledError`).

`cleanup_query` has been moved above `infos_cb` callback, as it should be the first
callback called.

libervia/backend/memory/disco.py

@@ -38,7 +38,7 @@ from libervia.backend.memory import persistent
 from tx_xmpp import disco
 import time
 
-from libervia.backend.tools.utils import ensure_deferred
+from libervia.backend.tools.utils import ensure_deferred, fork_deferred
 
 if TYPE_CHECKING:
     from libervia.backend.core.main import LiberviaBackend
@@ -296,7 +296,7 @@ class Discovery:
 
             if query_key in self._pending_queries:
                 log.debug(f"Reusing pending disco query for {jid_.full()}")
-                return self._pending_queries[query_key]
+                return fork_deferred(self._pending_queries[query_key])
 
             def cleanup_query(result):
                 """Remove the pending query on success or failure."""
@@ -312,7 +312,7 @@ class Discovery:
                 self._jid_disco_cache[jid_] = (disco_infos, time.time())
                 return disco_infos
 
-            def infos_eb(fail):
+            def infos_eb(fail: failure.Failure):
                 if fail.check(defer.CancelledError):
                     reason = "request time-out"
                     fail = failure.Failure(exceptions.TimeOutError(str(fail.value)))
@@ -340,11 +340,11 @@ class Discovery:
                 raise fail
 
             d = client.disco.requestInfo(jid_, nodeIdentifier=node)
-            d.addCallback(infos_cb)
             d.addBoth(cleanup_query)
+            d.addCallback(infos_cb)
             d.addErrback(infos_eb)
             self._pending_queries[query_key] = d
-            return d
+            return fork_deferred(d)
         else:
             disco_infos = self.hashes[cap_hash]
             return defer.succeed(disco_infos)
@@ -395,12 +395,11 @@ class Discovery:
 
         return items
 
-    def _infos_eb(self, failure_, entity_jid):
-        failure_.trap(StanzaError)
-        log.warning(
-            _("Error while requesting [%(jid)s]: %(error)s")
-            % {"jid": entity_jid.full(), "error": failure_.getErrorMessage()}
-        )
+    def _infos_eb(
+        self, failure_: failure.Failure, entity_jid: jid.JID
+    ) -> failure.Failure:
+        log.warning(f"Error while requesting [{entity_jid}]: {failure_}")
+        return failure_
 
     async def find_service_entity(
         self,

libervia/backend/tools/utils.py

@@ -18,7 +18,7 @@
 
 """various useful methods"""
 
-from typing import Any, Callable, Optional, Union
+from typing import Any, Callable, Optional, TypeVar, Union
 import unicodedata
 import os.path
 import datetime
@@ -168,6 +168,29 @@ def ensure_deferred(func):
     return wrapper
 
 
+_T = TypeVar("_T")
+
+
+def fork_deferred(source_d: defer.Deferred[_T]) -> defer.Deferred[_T]:
+    """Create an independent Deferred that mirrors the result of another.
+
+    Returns a new Deferred that fires with the same result (or failure) as ``source_d``,
+    without interfering with other callbacks on ``source``. The forked Deferred's
+    canceller will cancel ``source_d`` if not yet called.
+    """
+    new_d = defer.Deferred(lambda _: source_d.cancel() if not source_d.called else None)
+
+    def _propagate(result):
+        if isinstance(result, failure.Failure):
+            new_d.errback(result)
+        else:
+            new_d.callback(result)
+        return result
+
+    source_d.addBoth(_propagate)
+    return new_d
+
+
 def xmpp_date(
     timestamp: Optional[Union[float, int]] = None, with_time: bool = True
 ) -> str: