memory (disco): Fix error handling for find_service_entity/find_service_entities:

- Error raised by `disco_info` where not properly handler, resulting in an exception
  instead of appropriate logging and return value.
- `async` coroutines, type hints

libervia/backend/core/constants.py

@@ -18,7 +18,6 @@
 
 import os
 from pathlib import Path
-import tempfile
 from typing import Final
 
 from libervia import backend

libervia/backend/memory/disco.py

@@ -51,29 +51,27 @@ class HashGenerationError(Exception):
     pass
 
 
-class HashManager(object):
+class HashManager:
     """map object which manage hashes
 
     persistent storage is update when a new hash is added
     """
 
     def __init__(self, persistent):
-        self.hashes = {
-            CAP_HASH_ERROR: disco.DiscoInfo()  # used when we can't get disco infos
-        }
+        self.hashes: dict[str, disco.DiscoInfo] = {}
         self.persistent = persistent
 
-    def __getitem__(self, key):
+    def __getitem__(self, key) -> disco.DiscoInfo:
         return self.hashes[key]
 
-    def __setitem__(self, hash_, disco_info):
+    def __setitem__(self, hash_: str, disco_info: disco.DiscoInfo) -> None:
         if hash_ in self.hashes:
             log.debug("ignoring hash set: it is already known")
             return
         self.hashes[hash_] = disco_info
         self.persistent[hash_] = disco_info.toElement().toXml()
 
-    def __contains__(self, hash_):
+    def __contains__(self, hash_: str) -> bool:
         return self.hashes.__contains__(hash_)
 
     def load(self):
@@ -207,10 +205,14 @@ class Discovery:
     ) -> defer.Deferred[disco.DiscoInfo]:
         """get disco infos from jid_, filling capability hash if needed
 
+        @param client: client session
         @param jid_: jid of the target, or None for profile's server
         @param node: optional node to use for disco request
         @param use_cache: if True, use cached data if available
         @return: a Deferred which fire disco.DiscoInfo
+        @raise exceptions.ExternalRequestError: disco query failed and retry delay not
+            elapsed
+        @raise exceptions.TimeOutError: the request was cancelled
         """
         if jid_ is None:
             jid_ = jid.JID(client.jid.host)
@@ -233,10 +235,10 @@ class Discovery:
                             f"Cached disco failure for {jid_.full()}, retry in "
                             f"{DISCO_RETRY_DELAY - int(elapsed)} seconds"
                         )
-                        raise exceptions.ExternalRequestError(
+                        return defer.fail(exceptions.ExternalRequestError(
                             f"Disco query failed for {jid_.full()}, retrying in "
                             f"{DISCO_RETRY_DELAY} seconds"
-                        )
+                        ))
                     # Retry delay expired, remove from cache and retry
                     log.debug(f"Disco retry delay expired for {jid_.full()}, retrying")
                     del self._jid_disco_cache[jid_]
@@ -331,48 +333,63 @@ class Discovery:
             % {"jid": entity_jid.full(), "error": failure_.getErrorMessage()}
         )
 
-    def find_service_entity(self, client, category, type_, jid_=None):
+    async def find_service_entity(
+        self,
+        client: SatXMPPEntity,
+        category: str,
+        type_: str,
+        jid_: jid.JID | None = None,
+    ) -> jid.JID | None:
         """Helper method to find first available entity from find_service_entities
 
         args are the same as for [find_service_entities]
-        @return (jid.JID, None): found entity
+        @param client: client session
+        @param category: identity category
+        @param type_: identity type
+        @param jid_: jid of the target server, or None for profile's server
+        @return: first found entity or None
         """
-        d = self.host.find_service_entities(client, category, type_)
-        d.addCallback(lambda entities: entities.pop() if entities else None)
-        return d
+        entities = await self.host.find_service_entities(client, category, type_, jid_)
+        return entities.pop() if entities else None
 
-    def find_service_entities(self, client, category, type_, jid_=None):
+    async def find_service_entities(
+        self,
+        client: SatXMPPEntity,
+        category: str,
+        type_: str,
+        jid_: jid.JID | None = None,
+    ) -> set[jid.JID]:
         """Return all available items of an entity which correspond to (category, type_)
 
+        @param client: client session
         @param category: identity's category
         @param type_: identitiy's type
         @param jid_: the jid of the target server (None for profile's server)
         @return: a set of found entities
         @raise defer.CancelledError: the request timed out
         """
-        found_entities = set()
+        found_entities: set[jid.JID] = set()
 
         def infos_cb(infos, entity_jid):
             if (category, type_) in infos.identities:
                 found_entities.add(entity_jid)
 
-        def got_items(items):
-            defers_list = []
-            for item in items:
-                info_d = self.get_infos(client, item.entity)
-                info_d.addCallbacks(
-                    infos_cb, self._infos_eb, [item.entity], None, [item.entity]
-                )
-                defers_list.append(info_d)
-            return defer.DeferredList(defers_list)
+        items: disco.DiscoItems = await self.get_items(client, jid_)
+        defers_list = []
+        for item in items:
+            info_d = self.get_infos(client, item.entity)
 
-        d = self.get_items(client, jid_)
-        d.addCallback(got_items)
-        d.addCallback(lambda __: found_entities)
+
+            info_d.addCallbacks(
+                infos_cb, self._infos_eb, [item.entity], None, [item.entity]
+            )
+            defers_list.append(info_d)
+        d = defer.DeferredList(defers_list)
         reactor.callLater(
             TIMEOUT, d.cancel
         )  # FIXME: one bad service make a general timeout
-        return d
+        await d
+        return found_entities
 
     async def find_features_set(
         self,

libervia/backend/plugins/plugin_xep_0060.py

@@ -21,6 +21,7 @@ from collections import namedtuple
 from dataclasses import dataclass, field
 from functools import reduce
 from typing import (
+    TYPE_CHECKING,
     Any,
     Callable,
     ClassVar,
@@ -61,6 +62,9 @@ from libervia.backend.tools import sat_defer
 from libervia.backend.tools import xml_tools
 from libervia.backend.tools.common import data_format
 
+if TYPE_CHECKING:
+    from libervia.backend.core.main import LiberviaBackend
+
 log = getLogger(__name__)
 
 PLUGIN_INFO = {
@@ -263,7 +267,7 @@ class XEP_0060:
     # extra disco needed for RSM, cf. XEP-0060 § 6.5.4
     DISCO_RSM = "http://jabber.org/protocol/pubsub#rsm"
 
-    def __init__(self, host):
+    def __init__(self, host: "LiberviaBackend"):
         log.info(_("PubSub plugin initialization"))
         self.host = host
         self._rsm = cast(XEP_0059, host.plugins.get("XEP-0059"))

libervia/backend/plugins/plugin_xep_0277.py

@@ -1188,9 +1188,9 @@ class XEP_0277:
                     pass
                 else:
                     # other server, let's try to find a non PEP service there
-                    d = self.host.find_service_entity(
+                    d = defer.ensureDeferred(self.host.find_service_entity(
                         client, "pubsub", "service", parent_service
-                    )
+                    ))
                     d.addCallback(lambda entity: entity or parent_service)
             else:
                 # parent is already on a normal Pubsub service, we re-use it