changeset 3541:888109774673

core: various changes and fixes to work with new storage and D-Bus bridge: - fixes coroutines handling in various places - fixes types which are not serialised by Tx DBus - XEP-0384: call storage methods in main thread in XEP: Python OMEMO's Promise use thread which prevent the use of AsyncIO loop. To work around that, callLater is used to launch storage method in main thread. This is a temporary workaround, as Python OMEMO should get rid of Promise implementation and threads soon.
author Goffi <goffi@goffi.org>
date Thu, 03 Jun 2021 15:21:43 +0200 (2021-06-03)
parents aa58451b77ba
children 813595f88612
files sat/core/sat_main.py sat/core/xmpp.py sat/memory/memory.py sat/memory/persistent.py sat/plugins/plugin_comp_file_sharing.py sat/plugins/plugin_comp_file_sharing_management.py sat/plugins/plugin_misc_identity.py sat/plugins/plugin_xep_0033.py sat/plugins/plugin_xep_0045.py sat/plugins/plugin_xep_0198.py sat/plugins/plugin_xep_0313.py sat/plugins/plugin_xep_0329.py sat/plugins/plugin_xep_0384.py
diffstat 13 files changed, 221 insertions(+), 185 deletions(-) [+]
line wrap: on
line diff
--- a/sat/core/sat_main.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/core/sat_main.py	Thu Jun 03 15:21:43 2021 +0200
@@ -456,7 +456,7 @@
             except AttributeError:
                 continue
             else:
-                defers_list.append(defer.maybeDeferred(unload))
+                defers_list.append(utils.asDeferred(unload))
         return defers_list
 
     def _connect(self, profile_key, password="", options=None):
@@ -470,7 +470,7 @@
         Retrieve the individual parameters, authenticate the profile
         and initiate the connection to the associated XMPP server.
         @param profile: %(doc_profile)s
-        @param password (string): the SàT profile password
+        @param password (string): the Libervia profile password
         @param options (dict): connection options. Key can be:
             -
         @param max_retries (int): max number of connection retries
@@ -533,7 +533,7 @@
         features = []
         for import_name, plugin in self.plugins.items():
             try:
-                features_d = defer.maybeDeferred(plugin.getFeatures, profile_key)
+                features_d = utils.asDeferred(plugin.getFeatures, profile_key)
             except AttributeError:
                 features_d = defer.succeed({})
             features.append(features_d)
@@ -581,7 +581,7 @@
                 attr = client.roster.getAttributes(item)
                 # we use full() and not userhost() because jid with resources are allowed
                 # in roster, even if it's not common.
-                ret.append([item.entity.full(), attr, item.groups])
+                ret.append([item.entity.full(), attr, list(item.groups)])
             return ret
 
         return client.roster.got_roster.addCallback(got_roster)
@@ -1108,6 +1108,7 @@
     def _findByFeatures(self, namespaces, identities, bare_jids, service, roster, own_jid,
                         local_device, profile_key):
         client = self.getClient(profile_key)
+        identities = [tuple(i) for i in identities] if identities else None
         return defer.ensureDeferred(self.findByFeatures(
             client, namespaces, identities, bare_jids, service, roster, own_jid,
             local_device))
--- a/sat/core/xmpp.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/core/xmpp.py	Thu Jun 03 15:21:43 2021 +0200
@@ -714,7 +714,7 @@
             or mess_data["type"] == C.MESS_TYPE_INFO
         )
 
-    def messageAddToHistory(self, data):
+    async def messageAddToHistory(self, data):
         """Store message into database (for local history)
 
         @param data: message data dictionnary
@@ -726,7 +726,7 @@
 
             # we need a message to store
             if self.isMessagePrintable(data):
-                self.host_app.memory.addToHistory(self, data)
+                await self.host_app.memory.addToHistory(self, data)
             else:
                 log.warning(
                     "No message found"
@@ -876,7 +876,9 @@
 
     def addPostXmlCallbacks(self, post_xml_treatments):
         post_xml_treatments.addCallback(self.messageProt.completeAttachments)
-        post_xml_treatments.addCallback(self.messageAddToHistory)
+        post_xml_treatments.addCallback(
+            lambda ret: defer.ensureDeferred(self.messageAddToHistory(ret))
+        )
         post_xml_treatments.addCallback(self.messageSendToBridge)
 
     def send(self, obj):
@@ -1061,7 +1063,9 @@
 
     def addPostXmlCallbacks(self, post_xml_treatments):
         if self.sendHistory:
-            post_xml_treatments.addCallback(self.messageAddToHistory)
+            post_xml_treatments.addCallback(
+                lambda ret: defer.ensureDeferred(self.messageAddToHistory(ret))
+            )
 
     def getOwnerFromJid(self, to_jid: jid.JID) -> jid.JID:
         """Retrieve "owner" of a component resource from the destination jid of the request
@@ -1212,7 +1216,9 @@
         data = self.parseMessage(message_elt)
         post_treat.addCallback(self.completeAttachments)
         post_treat.addCallback(self.skipEmptyMessage)
-        post_treat.addCallback(self.addToHistory)
+        post_treat.addCallback(
+            lambda ret: defer.ensureDeferred(self.addToHistory(ret))
+        )
         post_treat.addCallback(self.bridgeSignal, data)
         post_treat.addErrback(self.cancelErrorTrap)
         post_treat.callback(data)
@@ -1253,14 +1259,14 @@
             raise failure.Failure(exceptions.CancelError("Cancelled empty message"))
         return data
 
-    def addToHistory(self, data):
+    async def addToHistory(self, data):
         if data.pop("history", None) == C.HISTORY_SKIP:
             log.debug("history is skipped as requested")
             data["extra"]["history"] = C.HISTORY_SKIP
         else:
             # we need a message to store
             if self.parent.isMessagePrintable(data):
-                return self.host.memory.addToHistory(self.parent, data)
+                return await self.host.memory.addToHistory(self.parent, data)
             else:
                 log.debug("not storing empty message to history: {data}"
                     .format(data=data))
@@ -1478,7 +1484,8 @@
         self._jids[entity] = item
         self._registerItem(item)
         self.host.bridge.newContact(
-            entity.full(), self.getAttributes(item), item.groups, self.parent.profile
+            entity.full(), self.getAttributes(item), list(item.groups),
+            self.parent.profile
         )
 
     def removeReceived(self, request):
--- a/sat/memory/memory.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/memory/memory.py	Thu Jun 03 15:21:43 2021 +0200
@@ -1131,12 +1131,12 @@
         )
 
     def asyncGetStringParamA(
-        self, name, category, attr="value", security_limit=C.NO_SECURITY_LIMIT,
+        self, name, category, attribute="value", security_limit=C.NO_SECURITY_LIMIT,
         profile_key=C.PROF_KEY_NONE):
 
         profile = self.getProfileName(profile_key)
         return defer.ensureDeferred(self.params.asyncGetStringParamA(
-            name, category, attr, security_limit, profile
+            name, category, attribute, security_limit, profile
         ))
 
     def _getParamsUI(self, security_limit, app, extra_s, profile_key):
@@ -1170,20 +1170,22 @@
         client = self.host.getClient(profile_key)
         # we accept any type
         data = data_format.deserialise(data_s, type_check=None)
-        return self.storage.setPrivateValue(
-            namespace, key, data, binary=True, profile=client.profile)
+        return defer.ensureDeferred(self.storage.setPrivateValue(
+            namespace, key, data, binary=True, profile=client.profile))
 
     def _privateDataGet(self, namespace, key, profile_key):
         client = self.host.getClient(profile_key)
-        d = self.storage.getPrivates(
-            namespace, [key], binary=True, profile=client.profile)
+        d = defer.ensureDeferred(
+            self.storage.getPrivates(
+                namespace, [key], binary=True, profile=client.profile)
+        )
         d.addCallback(lambda data_dict: data_format.serialise(data_dict.get(key)))
         return d
 
     def _privateDataDelete(self, namespace, key, profile_key):
         client = self.host.getClient(profile_key)
-        return self.storage.delPrivateValue(
-            namespace, key, binary=True, profile=client.profile)
+        return defer.ensureDeferred(self.storage.delPrivateValue(
+            namespace, key, binary=True, profile=client.profile))
 
     ## Files ##
 
@@ -1251,8 +1253,7 @@
                     _("unknown access type: {type}").format(type=perm_type)
                 )
 
-    @defer.inlineCallbacks
-    def checkPermissionToRoot(self, client, file_data, peer_jid, perms_to_check):
+    async def checkPermissionToRoot(self, client, file_data, peer_jid, perms_to_check):
         """do checkFilePermission on file_data and all its parents until root"""
         current = file_data
         while True:
@@ -1260,7 +1261,7 @@
             parent = current["parent"]
             if not parent:
                 break
-            files_data = yield self.getFiles(
+            files_data = await self.getFiles(
                 client, peer_jid=None, file_id=parent, perms_to_check=None
             )
             try:
@@ -1268,8 +1269,7 @@
             except IndexError:
                 raise exceptions.DataError("Missing parent")
 
-    @defer.inlineCallbacks
-    def _getParentDir(
+    async def _getParentDir(
         self, client, path, parent, namespace, owner, peer_jid, perms_to_check
     ):
         """Retrieve parent node from a path, or last existing directory
@@ -1293,7 +1293,7 @@
         # non existing directories will be created
         parent = ""
         for idx, path_elt in enumerate(path_elts):
-            directories = yield self.storage.getFiles(
+            directories = await self.storage.getFiles(
                 client,
                 parent=parent,
                 type_=C.FILE_TYPE_DIRECTORY,
@@ -1302,7 +1302,7 @@
                 owner=owner,
             )
             if not directories:
-                defer.returnValue((parent, path_elts[idx:]))
+                return (parent, path_elts[idx:])
                 # from this point, directories don't exist anymore, we have to create them
             elif len(directories) > 1:
                 raise exceptions.InternalError(
@@ -1312,7 +1312,7 @@
                 directory = directories[0]
                 self.checkFilePermission(directory, peer_jid, perms_to_check)
                 parent = directory["id"]
-        defer.returnValue((parent, []))
+        return (parent, [])
 
     def getFileAffiliations(self, file_data: dict) -> Dict[jid.JID, str]:
         """Convert file access to pubsub like affiliations"""
@@ -1484,8 +1484,7 @@
             )
         return peer_jid.userhostJID()
 
-    @defer.inlineCallbacks
-    def getFiles(
+    async def getFiles(
         self, client, peer_jid, file_id=None, version=None, parent=None, path=None,
         type_=None, file_hash=None, hash_algo=None, name=None, namespace=None,
         mime_type=None, public_id=None, owner=None, access=None, projection=None,
@@ -1536,7 +1535,7 @@
         if path is not None:
             path = str(path)
             # permission are checked by _getParentDir
-            parent, remaining_path_elts = yield self._getParentDir(
+            parent, remaining_path_elts = await self._getParentDir(
                 client, path, parent, namespace, owner, peer_jid, perms_to_check
             )
             if remaining_path_elts:
@@ -1546,16 +1545,16 @@
         if parent and peer_jid:
             # if parent is given directly and permission check is requested,
             # we need to check all the parents
-            parent_data = yield self.storage.getFiles(client, file_id=parent)
+            parent_data = await self.storage.getFiles(client, file_id=parent)
             try:
                 parent_data = parent_data[0]
             except IndexError:
                 raise exceptions.DataError("mising parent")
-            yield self.checkPermissionToRoot(
+            await self.checkPermissionToRoot(
                 client, parent_data, peer_jid, perms_to_check
             )
 
-        files = yield self.storage.getFiles(
+        files = await self.storage.getFiles(
             client,
             file_id=file_id,
             version=version,
@@ -1578,15 +1577,16 @@
             to_remove = []
             for file_data in files:
                 try:
-                    self.checkFilePermission(file_data, peer_jid, perms_to_check, set_affiliation=True)
+                    self.checkFilePermission(
+                        file_data, peer_jid, perms_to_check, set_affiliation=True
+                    )
                 except exceptions.PermissionError:
                     to_remove.append(file_data)
             for file_data in to_remove:
                 files.remove(file_data)
-        defer.returnValue(files)
+        return files
 
-    @defer.inlineCallbacks
-    def setFile(
+    async def setFile(
         self, client, name, file_id=None, version="", parent=None, path=None,
         type_=C.FILE_TYPE_FILE, file_hash=None, hash_algo=None, size=None,
         namespace=None, mime_type=None, public_id=None, created=None, modified=None,
@@ -1668,13 +1668,13 @@
         if path is not None:
             path = str(path)
             # _getParentDir will check permissions if peer_jid is set, so we use owner
-            parent, remaining_path_elts = yield self._getParentDir(
+            parent, remaining_path_elts = await self._getParentDir(
                 client, path, parent, namespace, owner, owner, perms_to_check
             )
             # if remaining directories don't exist, we have to create them
             for new_dir in remaining_path_elts:
                 new_dir_id = shortuuid.uuid()
-                yield self.storage.setFile(
+                await self.storage.setFile(
                     client,
                     name=new_dir,
                     file_id=new_dir_id,
@@ -1691,7 +1691,7 @@
         elif parent is None:
             parent = ""
 
-        yield self.storage.setFile(
+        await self.storage.setFile(
             client,
             file_id=file_id,
             version=version,
--- a/sat/memory/persistent.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/memory/persistent.py	Thu Jun 03 15:21:43 2021 +0200
@@ -17,10 +17,13 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+from twisted.internet import defer
+from twisted.python import failure
 from sat.core.i18n import _
 from sat.core.log import getLogger
+
+
 log = getLogger(__name__)
-from twisted.python import failure
 
 
 class MemoryNotInitializedError(Exception):
@@ -57,7 +60,9 @@
         need to be called before any other operation
         @return: defers the PersistentDict instance itself
         """
-        d = self.storage.getPrivates(self.namespace, binary=self.binary, profile=self.profile)
+        d = defer.ensureDeferred(self.storage.getPrivates(
+            self.namespace, binary=self.binary, profile=self.profile
+        ))
         d.addCallback(self._setCache)
         d.addCallback(lambda __: self)
         return d
@@ -111,8 +116,11 @@
         return self._cache.__getitem__(key)
 
     def __setitem__(self, key, value):
-        self.storage.setPrivateValue(self.namespace, key, value, self.binary,
-                                     self.profile)
+        defer.ensureDeferred(
+            self.storage.setPrivateValue(
+                self.namespace, key, value, self.binary, self.profile
+            )
+        )
         return self._cache.__setitem__(key, value)
 
     def __delitem__(self, key):
@@ -130,8 +138,11 @@
     def aset(self, key, value):
         """Async set, return a Deferred fired when value is actually stored"""
         self._cache.__setitem__(key, value)
-        return self.storage.setPrivateValue(self.namespace, key, value,
-                                            self.binary, self.profile)
+        return defer.ensureDeferred(
+            self.storage.setPrivateValue(
+                self.namespace, key, value, self.binary, self.profile
+            )
+        )
 
     def adel(self, key):
         """Async del, return a Deferred fired when value is actually deleted"""
@@ -151,8 +162,11 @@
 
         @return: deferred fired when data is actually saved
         """
-        return self.storage.setPrivateValue(self.namespace, name, self._cache[name],
-                                            self.binary, self.profile)
+        return defer.ensureDeferred(
+            self.storage.setPrivateValue(
+                self.namespace, name, self._cache[name], self.binary, self.profile
+            )
+        )
 
 
 class PersistentBinaryDict(PersistentDict):
@@ -178,12 +192,16 @@
         raise NotImplementedError
 
     def items(self):
-        d = self.storage.getPrivates(self.namespace, binary=self.binary, profile=self.profile)
+        d = defer.ensureDeferred(self.storage.getPrivates(
+            self.namespace, binary=self.binary, profile=self.profile
+        ))
         d.addCallback(lambda data_dict: data_dict.items())
         return d
 
     def all(self):
-        return self.storage.getPrivates(self.namespace, binary=self.binary, profile=self.profile)
+        return defer.ensureDeferred(self.storage.getPrivates(
+            self.namespace, binary=self.binary, profile=self.profile
+        ))
 
     def __repr__(self):
         raise NotImplementedError
@@ -234,14 +252,18 @@
 
     def __getitem__(self, key):
         """get the value as a Deferred"""
-        d = self.storage.getPrivates(self.namespace, keys=[key], binary=self.binary,
-                                     profile=self.profile)
+        d = defer.ensureDeferred(self.storage.getPrivates(
+            self.namespace, keys=[key], binary=self.binary, profile=self.profile
+        ))
         d.addCallback(self._data2value, key)
         return d
 
     def __setitem__(self, key, value):
-        self.storage.setPrivateValue(self.namespace, key, value, self.binary,
-                                     self.profile)
+        defer.ensureDeferred(
+            self.storage.setPrivateValue(
+                self.namespace, key, value, self.binary, self.profile
+            )
+        )
 
     def __delitem__(self, key):
         self.storage.delPrivateValue(self.namespace, key, self.binary, self.profile)
@@ -259,8 +281,11 @@
         """Async set, return a Deferred fired when value is actually stored"""
         # FIXME: redundant with force, force must be removed
         # XXX: similar as PersistentDict.aset, but doesn't use cache
-        return self.storage.setPrivateValue(self.namespace, key, value,
-                                            self.binary, self.profile)
+        return defer.ensureDeferred(
+            self.storage.setPrivateValue(
+                self.namespace, key, value, self.binary, self.profile
+            )
+        )
 
     def adel(self, key):
         """Async del, return a Deferred fired when value is actually deleted"""
@@ -277,7 +302,11 @@
         @param value(object): value is needed for LazyPersistentBinaryDict
         @return: deferred fired when data is actually saved
         """
-        return self.storage.setPrivateValue(self.namespace, name, value, self.binary, self.profile)
+        return defer.ensureDeferred(
+            self.storage.setPrivateValue(
+                self.namespace, name, value, self.binary, self.profile
+            )
+        )
 
     def remove(self, key):
         """Delete a key from sotrage, and return a deferred called when it's done
--- a/sat/plugins/plugin_comp_file_sharing.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_comp_file_sharing.py	Thu Jun 03 15:21:43 2021 +0200
@@ -31,6 +31,7 @@
 from sat.core.log import getLogger
 from sat.tools import stream
 from sat.tools import video
+from sat.tools.utils import ensure_deferred
 from sat.tools.common import regex
 from sat.tools.common import uri
 from sat.tools.common import files_utils
@@ -487,7 +488,7 @@
                 else:
                     await self.generate_thumbnails(extra, thumb_path)
 
-        self.host.memory.setFile(
+        await self.host.memory.setFile(
             client,
             name=name,
             version="",
@@ -546,8 +547,7 @@
         )
         return False, defer.succeed(True)
 
-    @defer.inlineCallbacks
-    def _retrieveFiles(
+    async def _retrieveFiles(
         self, client, session, content_data, content_name, file_data, file_elt
     ):
         """This method retrieve a file on request, and send if after checking permissions"""
@@ -557,7 +557,7 @@
         else:
             owner = peer_jid
         try:
-            found_files = yield self.host.memory.getFiles(
+            found_files = await self.host.memory.getFiles(
                 client,
                 peer_jid=peer_jid,
                 name=file_data.get("name"),
@@ -575,13 +575,13 @@
                     peer_jid=peer_jid, name=file_data.get("name")
                 )
             )
-            defer.returnValue(False)
+            return False
 
         if not found_files:
             log.warning(
                 _("no matching file found ({file_data})").format(file_data=file_data)
             )
-            defer.returnValue(False)
+            return False
 
         # we only use the first found file
         found_file = found_files[0]
@@ -607,7 +607,7 @@
             size=size,
             data_cb=lambda data: hasher.update(data),
         )
-        defer.returnValue(True)
+        return True
 
     def _fileSendingRequestTrigger(
         self, client, session, content_data, content_name, file_data, file_elt
@@ -617,9 +617,9 @@
         else:
             return (
                 False,
-                self._retrieveFiles(
+                defer.ensureDeferred(self._retrieveFiles(
                     client, session, content_data, content_name, file_data, file_elt
-                ),
+                )),
             )
 
     ## HTTP Upload ##
@@ -757,11 +757,10 @@
             raise error.StanzaError("item-not-found")
         return file_id
 
-    @defer.inlineCallbacks
-    def getFileData(self, requestor, nodeIdentifier):
+    async def getFileData(self, requestor, nodeIdentifier):
         file_id = self._getFileId(nodeIdentifier)
         try:
-            files = yield self.host.memory.getFiles(self.parent, requestor, file_id)
+            files = await self.host.memory.getFiles(self.parent, requestor, file_id)
         except (exceptions.NotFound, exceptions.PermissionError):
             # we don't differenciate between NotFound and PermissionError
             # to avoid leaking information on existing files
@@ -770,7 +769,7 @@
             raise error.StanzaError("item-not-found")
         if len(files) > 1:
             raise error.InternalError("there should be only one file")
-        defer.returnValue(files[0])
+        return files[0]
 
     def commentsUpdate(self, extra, new_comments, peer_jid):
         """update comments (replace or insert new_comments)
@@ -825,10 +824,10 @@
             iq_elt = iq_elt.parent
         return iq_elt["from"]
 
-    @defer.inlineCallbacks
-    def publish(self, requestor, service, nodeIdentifier, items):
+    @ensure_deferred
+    async def publish(self, requestor, service, nodeIdentifier, items):
         #  we retrieve file a first time to check authorisations
-        file_data = yield self.getFileData(requestor, nodeIdentifier)
+        file_data = await self.getFileData(requestor, nodeIdentifier)
         file_id = file_data["id"]
         comments = [(item["id"], self._getFrom(item), item.toXml()) for item in items]
         if requestor.userhostJID() == file_data["owner"]:
@@ -837,24 +836,24 @@
             peer_jid = requestor.userhost()
         update_cb = partial(self.commentsUpdate, new_comments=comments, peer_jid=peer_jid)
         try:
-            yield self.host.memory.fileUpdate(file_id, "extra", update_cb)
+            await self.host.memory.fileUpdate(file_id, "extra", update_cb)
         except exceptions.PermissionError:
             raise error.StanzaError("not-authorized")
 
-    @defer.inlineCallbacks
-    def items(self, requestor, service, nodeIdentifier, maxItems, itemIdentifiers):
-        file_data = yield self.getFileData(requestor, nodeIdentifier)
+    @ensure_deferred
+    async def items(self, requestor, service, nodeIdentifier, maxItems, itemIdentifiers):
+        file_data = await self.getFileData(requestor, nodeIdentifier)
         comments = file_data["extra"].get("comments", [])
         if itemIdentifiers:
             defer.returnValue(
                 [generic.parseXml(c[2]) for c in comments if c[0] in itemIdentifiers]
             )
         else:
-            defer.returnValue([generic.parseXml(c[2]) for c in comments])
+            return [generic.parseXml(c[2]) for c in comments]
 
-    @defer.inlineCallbacks
-    def retract(self, requestor, service, nodeIdentifier, itemIdentifiers):
-        file_data = yield self.getFileData(requestor, nodeIdentifier)
+    @ensure_deferred
+    async def retract(self, requestor, service, nodeIdentifier, itemIdentifiers):
+        file_data = await self.getFileData(requestor, nodeIdentifier)
         file_id = file_data["id"]
         try:
             comments = file_data["extra"]["comments"]
@@ -879,4 +878,4 @@
                 raise error.StanzaError("not-authorized")
 
         remove_cb = partial(self.commentsDelete, comments=to_remove)
-        yield self.host.memory.fileUpdate(file_id, "extra", remove_cb)
+        await self.host.memory.fileUpdate(file_id, "extra", remove_cb)
--- a/sat/plugins/plugin_comp_file_sharing_management.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_comp_file_sharing_management.py	Thu Jun 03 15:21:43 2021 +0200
@@ -149,8 +149,7 @@
         payload = form.toElement()
         return payload, status, None, None
 
-    @defer.inlineCallbacks
-    def _getFileData(self, client, session_data, command_form):
+    async def _getFileData(self, client, session_data, command_form):
         """Retrieve field requested in root form
 
         "found_file" will also be set in session_data
@@ -177,7 +176,7 @@
         #       this must be managed
 
         try:
-            found_files = yield self.host.memory.getFiles(
+            found_files = await self.host.memory.getFiles(
                 client, requestor_bare, path=parent_path, name=basename,
                 namespace=namespace)
             found_file = found_files[0]
@@ -193,7 +192,7 @@
 
         session_data['found_file'] = found_file
         session_data['namespace'] = namespace
-        defer.returnValue(found_file)
+        return found_file
 
     def _updateReadPermission(self, access, allowed_jids):
         if not allowed_jids:
@@ -209,29 +208,27 @@
                 "jids": [j.full() for j in allowed_jids]
             }
 
-    @defer.inlineCallbacks
-    def _updateDir(self, client, requestor, namespace, file_data, allowed_jids):
+    async def _updateDir(self, client, requestor, namespace, file_data, allowed_jids):
         """Recursively update permission of a directory and all subdirectories
 
         @param file_data(dict): metadata of the file
         @param allowed_jids(list[jid.JID]): list of entities allowed to read the file
         """
         assert file_data['type'] == C.FILE_TYPE_DIRECTORY
-        files_data = yield self.host.memory.getFiles(
+        files_data = await self.host.memory.getFiles(
             client, requestor, parent=file_data['id'], namespace=namespace)
 
         for file_data in files_data:
             if not file_data['access'].get(C.ACCESS_PERM_READ, {}):
                 log.debug("setting {perm} read permission for {name}".format(
                     perm=allowed_jids, name=file_data['name']))
-                yield self.host.memory.fileUpdate(
+                await self.host.memory.fileUpdate(
                     file_data['id'], 'access',
                     partial(self._updateReadPermission, allowed_jids=allowed_jids))
             if file_data['type'] == C.FILE_TYPE_DIRECTORY:
-                yield self._updateDir(client, requestor, namespace, file_data, 'PUBLIC')
+                await self._updateDir(client, requestor, namespace, file_data, 'PUBLIC')
 
-    @defer.inlineCallbacks
-    def _onChangeFile(self, client, command_elt, session_data, action, node):
+    async def _onChangeFile(self, client, command_elt, session_data, action, node):
         try:
             x_elt = next(command_elt.elements(data_form.NS_X_DATA, "x"))
             command_form = data_form.Form.fromElement(x_elt)
@@ -244,14 +241,14 @@
 
         if command_form is None or len(command_form.fields) == 0:
             # root request
-            defer.returnValue(self._getRootArgs())
+            return self._getRootArgs()
 
         elif found_file is None:
             # file selected, we retrieve it and ask for permissions
             try:
-                found_file = yield self._getFileData(client, session_data, command_form)
+                found_file = await self._getFileData(client, session_data, command_form)
             except WorkflowError as e:
-                defer.returnValue(e.err_args)
+                return e.err_args
 
             # management request
             if found_file['type'] == C.FILE_TYPE_DIRECTORY:
@@ -284,7 +281,7 @@
 
             status = self._c.STATUS.EXECUTING
             payload = form.toElement()
-            defer.returnValue((payload, status, None, None))
+            return (payload, status, None, None)
 
         else:
             # final phase, we'll do permission change here
@@ -307,7 +304,7 @@
                     self._c.adHocError(self._c.ERROR.BAD_PAYLOAD)
 
             if found_file['type'] == C.FILE_TYPE_FILE:
-                yield self.host.memory.fileUpdate(
+                await self.host.memory.fileUpdate(
                     found_file['id'], 'access',
                     partial(self._updateReadPermission, allowed_jids=allowed_jids))
             else:
@@ -315,7 +312,7 @@
                     recursive = command_form.fields['recursive']
                 except KeyError:
                     self._c.adHocError(self._c.ERROR.BAD_PAYLOAD)
-                yield self.host.memory.fileUpdate(
+                await self.host.memory.fileUpdate(
                     found_file['id'], 'access',
                     partial(self._updateReadPermission, allowed_jids=allowed_jids))
                 if recursive:
@@ -323,17 +320,16 @@
                     # already a permission set), so allowed entities of root directory
                     # can read them.
                     namespace = session_data['namespace']
-                    yield self._updateDir(
+                    await self._updateDir(
                         client, requestor_bare, namespace, found_file, 'PUBLIC')
 
             # job done, we can end the session
             status = self._c.STATUS.COMPLETED
             payload = None
             note = (self._c.NOTE.INFO, _("management session done"))
-            defer.returnValue((payload, status, None, note))
+            return (payload, status, None, note)
 
-    @defer.inlineCallbacks
-    def _onDeleteFile(self, client, command_elt, session_data, action, node):
+    async def _onDeleteFile(self, client, command_elt, session_data, action, node):
         try:
             x_elt = next(command_elt.elements(data_form.NS_X_DATA, "x"))
             command_form = data_form.Form.fromElement(x_elt)
@@ -346,14 +342,14 @@
 
         if command_form is None or len(command_form.fields) == 0:
             # root request
-            defer.returnValue(self._getRootArgs())
+            return self._getRootArgs()
 
         elif found_file is None:
             # file selected, we need confirmation before actually deleting
             try:
-                found_file = yield self._getFileData(client, session_data, command_form)
+                found_file = await self._getFileData(client, session_data, command_form)
             except WorkflowError as e:
-                defer.returnValue(e.err_args)
+                return e.err_args
             if found_file['type'] == C.FILE_TYPE_DIRECTORY:
                 msg = D_("Are you sure to delete directory {name} and all files and "
                          "directories under it?").format(name=found_file['name'])
@@ -370,7 +366,7 @@
             form.addField(field)
             status = self._c.STATUS.EXECUTING
             payload = form.toElement()
-            defer.returnValue((payload, status, None, None))
+            return (payload, status, None, None)
 
         else:
             # final phase, we'll do deletion here
@@ -382,27 +378,26 @@
                 note = None
             else:
                 recursive = found_file['type'] == C.FILE_TYPE_DIRECTORY
-                yield self.host.memory.fileDelete(
+                await self.host.memory.fileDelete(
                     client, requestor_bare, found_file['id'], recursive)
                 note = (self._c.NOTE.INFO, _("file deleted"))
             status = self._c.STATUS.COMPLETED
             payload = None
-            defer.returnValue((payload, status, None, note))
+            return (payload, status, None, note)
 
     def _updateThumbs(self, extra, thumbnails):
         extra[C.KEY_THUMBNAILS] = thumbnails
 
-    @defer.inlineCallbacks
-    def _genThumbs(self, client, requestor, namespace, file_data):
+    async def _genThumbs(self, client, requestor, namespace, file_data):
         """Recursively generate thumbnails
 
         @param file_data(dict): metadata of the file
         """
         if file_data['type'] == C.FILE_TYPE_DIRECTORY:
-            sub_files_data = yield self.host.memory.getFiles(
+            sub_files_data = await self.host.memory.getFiles(
                 client, requestor, parent=file_data['id'], namespace=namespace)
             for sub_file_data in sub_files_data:
-                yield self._genThumbs(client, requestor, namespace, sub_file_data)
+                await self._genThumbs(client, requestor, namespace, sub_file_data)
 
         elif file_data['type'] == C.FILE_TYPE_FILE:
             media_type = file_data['media_type']
@@ -412,7 +407,7 @@
 
                 for max_thumb_size in self._t.SIZES:
                     try:
-                        thumb_size, thumb_id = yield self._t.generateThumbnail(
+                        thumb_size, thumb_id = await self._t.generateThumbnail(
                             file_path,
                             max_thumb_size,
                             #  we keep thumbnails for 6 months
@@ -424,7 +419,7 @@
                         break
                     thumbnails.append({"id": thumb_id, "size": thumb_size})
 
-                yield self.host.memory.fileUpdate(
+                await self.host.memory.fileUpdate(
                     file_data['id'], 'extra',
                     partial(self._updateThumbs, thumbnails=thumbnails))
 
@@ -434,8 +429,7 @@
         else:
             log.warning("unmanaged file type: {type_}".format(type_=file_data['type']))
 
-    @defer.inlineCallbacks
-    def _onGenThumbnails(self, client, command_elt, session_data, action, node):
+    async def _onGenThumbnails(self, client, command_elt, session_data, action, node):
         try:
             x_elt = next(command_elt.elements(data_form.NS_X_DATA, "x"))
             command_form = data_form.Form.fromElement(x_elt)
@@ -447,23 +441,23 @@
 
         if command_form is None or len(command_form.fields) == 0:
             # root request
-            defer.returnValue(self._getRootArgs())
+            return self._getRootArgs()
 
         elif found_file is None:
             # file selected, we retrieve it and ask for permissions
             try:
-                found_file = yield self._getFileData(client, session_data, command_form)
+                found_file = await self._getFileData(client, session_data, command_form)
             except WorkflowError as e:
-                defer.returnValue(e.err_args)
+                return e.err_args
 
             log.info("Generating thumbnails as requested")
-            yield self._genThumbs(client, requestor, found_file['namespace'], found_file)
+            await self._genThumbs(client, requestor, found_file['namespace'], found_file)
 
             # job done, we can end the session
             status = self._c.STATUS.COMPLETED
             payload = None
             note = (self._c.NOTE.INFO, _("thumbnails generated"))
-            defer.returnValue((payload, status, None, note))
+            return (payload, status, None, note)
 
     async def _onQuota(self, client, command_elt, session_data, action, node):
         requestor = session_data['requestor']
--- a/sat/plugins/plugin_misc_identity.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_misc_identity.py	Thu Jun 03 15:21:43 2021 +0200
@@ -142,9 +142,6 @@
 
         stored_data = await client._identity_storage.all()
 
-        self.host.memory.storage.getPrivates(
-            namespace="identity", binary=True, profile=client.profile)
-
         to_delete = []
 
         for key, value in stored_data.items():
--- a/sat/plugins/plugin_xep_0033.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_xep_0033.py	Thu Jun 03 15:21:43 2021 +0200
@@ -156,7 +156,9 @@
             d = defer.Deferred()
             if not skip_send:
                 d.addCallback(client.sendMessageData)
-            d.addCallback(client.messageAddToHistory)
+            d.addCallback(
+                lambda ret: defer.ensureDeferred(client.messageAddToHistory(ret))
+            )
             d.addCallback(client.messageSendToBridge)
             d.addErrback(lambda failure: failure.trap(exceptions.CancelError))
             return d.callback(mess_data)
--- a/sat/plugins/plugin_xep_0045.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_xep_0045.py	Thu Jun 03 15:21:43 2021 +0200
@@ -837,7 +837,8 @@
             elif client.muc_service is not None:
                 service = client.muc_service
             else:
-                msg = D_("No known default MUC service".format(unparsed))
+                msg = D_("No known default MUC service {unparsed}").format(
+                    unparsed=unparsed)
                 self.text_cmds.feedBack(client, msg, mess_data)
                 return False
         except jid.InvalidFormat:
@@ -1298,7 +1299,9 @@
         except AttributeError:
             mess_data = self.client.messageProt.parseMessage(message.element)
         if mess_data['message'] or mess_data['subject']:
-            return self.host.memory.addToHistory(self.client, mess_data)
+            return defer.ensureDeferred(
+                self.host.memory.addToHistory(self.client, mess_data)
+            )
         else:
             return defer.succeed(None)
 
--- a/sat/plugins/plugin_xep_0198.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_xep_0198.py	Thu Jun 03 15:21:43 2021 +0200
@@ -448,7 +448,7 @@
             d.addCallback(lambda __: client.roster.got_roster)
             if plg_0313 is not None:
                 # we retrieve one2one MAM archives
-                d.addCallback(lambda __: plg_0313.resume(client))
+                d.addCallback(lambda __: defer.ensureDeferred(plg_0313.resume(client)))
             # initial presence must be sent manually
             d.addCallback(lambda __: client.presence.available())
             if plg_0045 is not None:
--- a/sat/plugins/plugin_xep_0313.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_xep_0313.py	Thu Jun 03 15:21:43 2021 +0200
@@ -75,15 +75,14 @@
             out_sign='(a(sdssa{ss}a{ss}ss)ss)', method=self._getArchives,
             async_=True)
 
-    @defer.inlineCallbacks
-    def resume(self, client):
+    async def resume(self, client):
         """Retrieve one2one messages received since the last we have in local storage"""
-        stanza_id_data = yield self.host.memory.storage.getPrivates(
+        stanza_id_data = await self.host.memory.storage.getPrivates(
             mam.NS_MAM, [KEY_LAST_STANZA_ID], profile=client.profile)
         stanza_id = stanza_id_data.get(KEY_LAST_STANZA_ID)
         if stanza_id is None:
             log.info("can't retrieve last stanza ID, checking history")
-            last_mess = yield self.host.memory.historyGet(
+            last_mess = await self.host.memory.historyGet(
                 None, None, limit=1, filters={'not_types': C.MESS_TYPE_GROUPCHAT,
                                               'last_stanza_id': True},
                 profile=client.profile)
@@ -98,7 +97,7 @@
         complete = False
         count = 0
         while not complete:
-            mam_data = yield self.getArchives(client, mam_req,
+            mam_data = await self.getArchives(client, mam_req,
                                               service=client.jid.userhostJID())
             elt_list, rsm_response, mam_response = mam_data
             complete = mam_response["complete"]
@@ -141,7 +140,7 @@
                     # adding message to history
                     mess_data = client.messageProt.parseMessage(fwd_message_elt)
                     try:
-                        yield client.messageProt.addToHistory(mess_data)
+                        await client.messageProt.addToHistory(mess_data)
                     except exceptions.CancelError as e:
                         log.warning(
                             "message has not been added to history: {e}".format(e=e))
@@ -156,8 +155,8 @@
             log.info(_("We have received {num_mess} message(s) while offline.")
                 .format(num_mess=count))
 
-    def profileConnected(self, client):
-        return self.resume(client)
+    async def profileConnected(self, client):
+        await self.resume(client)
 
     def getHandler(self, client):
         mam_client = client._mam = SatMAMClient(self)
--- a/sat/plugins/plugin_xep_0329.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_xep_0329.py	Thu Jun 03 15:21:43 2021 +0200
@@ -19,6 +19,7 @@
 import mimetypes
 import json
 import os
+import traceback
 from pathlib import Path
 from typing import Optional, Dict
 from zope.interface import implementer
@@ -33,6 +34,7 @@
 from sat.core.constants import Const as C
 from sat.core.log import getLogger
 from sat.tools import stream
+from sat.tools import utils
 from sat.tools.common import regex
 
 
@@ -453,9 +455,9 @@
         iq_elt.handled = True
         node = iq_elt.query.getAttribute("node")
         if not node:
-            d = defer.maybeDeferred(root_nodes_cb, client, iq_elt)
+            d = utils.asDeferred(root_nodes_cb, client, iq_elt)
         else:
-            d = defer.maybeDeferred(files_from_node_cb, client, iq_elt, node)
+            d = utils.asDeferred(files_from_node_cb, client, iq_elt, node)
         d.addErrback(
             lambda failure_: log.error(
                 _("error while retrieving files: {msg}").format(msg=failure_)
@@ -589,10 +591,9 @@
         @return (tuple[jid.JID, jid.JID]): peer_jid and owner
         """
 
-    @defer.inlineCallbacks
-    def _compGetRootNodesCb(self, client, iq_elt):
+    async def _compGetRootNodesCb(self, client, iq_elt):
         peer_jid, owner = client.getOwnerAndPeer(iq_elt)
-        files_data = yield self.host.memory.getFiles(
+        files_data = await self.host.memory.getFiles(
             client,
             peer_jid=peer_jid,
             parent="",
@@ -607,8 +608,7 @@
             directory_elt["name"] = name
         client.send(iq_result_elt)
 
-    @defer.inlineCallbacks
-    def _compGetFilesFromNodeCb(self, client, iq_elt, node_path):
+    async def _compGetFilesFromNodeCb(self, client, iq_elt, node_path):
         """Retrieve files from local files repository according to permissions
 
         result stanza is then built and sent to requestor
@@ -618,7 +618,7 @@
         """
         peer_jid, owner = client.getOwnerAndPeer(iq_elt)
         try:
-            files_data = yield self.host.memory.getFiles(
+            files_data = await self.host.memory.getFiles(
                 client, peer_jid=peer_jid, path=node_path, owner=owner
             )
         except exceptions.NotFound:
@@ -628,7 +628,8 @@
             self._iqError(client, iq_elt, condition='not-allowed')
             return
         except Exception as e:
-            log.error("internal server error: {e}".format(e=e))
+            tb = traceback.format_tb(e.__traceback__)
+            log.error(f"internal server error: {e}\n{''.join(tb)}")
             self._iqError(client, iq_elt, condition='internal-server-error')
             return
         iq_result_elt = xmlstream.toResponse(iq_elt, "result")
--- a/sat/plugins/plugin_xep_0384.py	Thu Jun 03 15:21:43 2021 +0200
+++ b/sat/plugins/plugin_xep_0384.py	Thu Jun 03 15:21:43 2021 +0200
@@ -114,7 +114,10 @@
     @return (defer.Deferred): deferred instance linked to the promise
     """
     d = defer.Deferred()
-    promise_.then(d.callback, d.errback)
+    promise_.then(
+        lambda result: reactor.callLater(0, d.callback, result),
+        lambda exc: reactor.callLater(0, d.errback, exc)
+    )
     return d
 
 
@@ -141,6 +144,26 @@
         deferred.addCallback(partial(callback, True))
         deferred.addErrback(partial(callback, False))
 
+    def _callMainThread(self, callback, method, *args, check_jid=None):
+        d = method(*args)
+        if check_jid is not None:
+            check_jid_d = self._checkJid(check_jid)
+            check_jid_d.addCallback(lambda __: d)
+            d = check_jid_d
+        if callback is not None:
+            d.addCallback(partial(callback, True))
+            d.addErrback(partial(callback, False))
+
+    def _call(self, callback, method, *args, check_jid=None):
+        """Create Deferred and add Promise callback to it
+
+        This method use reactor.callLater to launch Deferred in main thread
+        @param check_jid: run self._checkJid before method
+        """
+        reactor.callLater(
+            0, self._callMainThread, callback, method, *args, check_jid=check_jid
+        )
+
     def _checkJid(self, bare_jid):
         """Check if jid is known, and store it if not
 
@@ -164,71 +187,50 @@
         callback(True, None)
 
     def loadState(self, callback):
-        d = self.data.get(KEY_STATE)
-        self.setCb(d, callback)
+        self._call(callback, self.data.get, KEY_STATE)
 
     def storeState(self, callback, state):
-        d = self.data.force(KEY_STATE, state)
-        self.setCb(d, callback)
+        self._call(callback, self.data.force, KEY_STATE, state)
 
     def loadSession(self, callback, bare_jid, device_id):
         key = '\n'.join([KEY_SESSION, bare_jid, str(device_id)])
-        d = self.data.get(key)
-        self.setCb(d, callback)
+        self._call(callback, self.data.get, key)
 
     def storeSession(self, callback, bare_jid, device_id, session):
         key = '\n'.join([KEY_SESSION, bare_jid, str(device_id)])
-        d = self.data.force(key, session)
-        self.setCb(d, callback)
+        self._call(callback, self._data.force, key, session)
 
     def deleteSession(self, callback, bare_jid, device_id):
         key = '\n'.join([KEY_SESSION, bare_jid, str(device_id)])
-        d = self.data.remove(key)
-        self.setCb(d, callback)
+        self._call(callback, self.data.remove, key)
 
     def loadActiveDevices(self, callback, bare_jid):
         key = '\n'.join([KEY_ACTIVE_DEVICES, bare_jid])
-        d = self.data.get(key, {})
-        if callback is not None:
-            self.setCb(d, callback)
-        return d
+        self._call(callback, self.data.get, key, {})
 
     def loadInactiveDevices(self, callback, bare_jid):
         key = '\n'.join([KEY_INACTIVE_DEVICES, bare_jid])
-        d = self.data.get(key, {})
-        if callback is not None:
-            self.setCb(d, callback)
-        return d
+        self._call(callback, self.data.get, key, {})
 
     def storeActiveDevices(self, callback, bare_jid, devices):
         key = '\n'.join([KEY_ACTIVE_DEVICES, bare_jid])
-        d = self._checkJid(bare_jid)
-        d.addCallback(lambda _: self.data.force(key, devices))
-        self.setCb(d, callback)
+        self._call(callback, self.data.force, key, devices, check_jid=bare_jid)
 
     def storeInactiveDevices(self, callback, bare_jid, devices):
         key = '\n'.join([KEY_INACTIVE_DEVICES, bare_jid])
-        d = self._checkJid(bare_jid)
-        d.addCallback(lambda _: self.data.force(key, devices))
-        self.setCb(d, callback)
+        self._call(callback, self.data.force, key, devices, check_jid=bare_jid)
 
     def storeTrust(self, callback, bare_jid, device_id, trust):
         key = '\n'.join([KEY_TRUST, bare_jid, str(device_id)])
-        d = self.data.force(key, trust)
-        self.setCb(d, callback)
+        self._call(callback, self.data.force, key, trust)
 
     def loadTrust(self, callback, bare_jid, device_id):
         key = '\n'.join([KEY_TRUST, bare_jid, str(device_id)])
-        d = self.data.get(key)
-        if callback is not None:
-            self.setCb(d, callback)
-        return d
+        self._call(callback, self.data.get, key)
 
     def listJIDs(self, callback):
-        d = defer.succeed(self.all_jids)
         if callback is not None:
-            self.setCb(d, callback)
-        return d
+            callback(True, self.all_jids)
 
     def _deleteJID_logResults(self, results):
         failed = [success for success, __ in results if not success]
@@ -266,8 +268,7 @@
         d.addCallback(self._deleteJID_logResults)
         return d
 
-    def deleteJID(self, callback, bare_jid):
-        """Retrieve all (in)actives devices of bare_jid, and delete all related keys"""
+    def _deleteJID(self, callback, bare_jid):
         d_list = []
 
         key = '\n'.join([KEY_ACTIVE_DEVICES, bare_jid])
@@ -284,7 +285,10 @@
         d.addCallback(self._deleteJID_gotDevices, bare_jid)
         if callback is not None:
             self.setCb(d, callback)
-        return d
+
+    def deleteJID(self, callback, bare_jid):
+        """Retrieve all (in)actives devices of bare_jid, and delete all related keys"""
+        reactor.callLater(0, self._deleteJID, callback, bare_jid)
 
 
 class SatOTPKPolicy(omemo.DefaultOTPKPolicy):
@@ -728,7 +732,7 @@
             while device_id in devices:
                 device_id = random.randint(1, 2**31-1)
             # and we save it
-            persistent_dict[KEY_DEVICE_ID] = device_id
+            await persistent_dict.aset(KEY_DEVICE_ID, device_id)
 
         log.debug(f"our OMEMO device id is {device_id}")