Add comments

4 files changed, 61 insertions, 7 deletions

diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py
index 2d66b3721a..a2ffd273bf 100644
--- a/synapse/handlers/device.py
+++ b/synapse/handlers/device.py
@@ -192,6 +192,9 @@ class DeviceHandler(BaseHandler):
 
     @defer.inlineCallbacks
     def notify_device_update(self, user_id, device_ids):
+        """Notify that a user's device(s) has changed. Pokes the notifier, and
+        remote servers if the user is local.
+        """
         rooms = yield self.store.get_rooms_for_user(user_id)
         room_ids = [r.room_id for r in rooms]
 
@@ -210,12 +213,16 @@ class DeviceHandler(BaseHandler):
             "device_list_key", position, rooms=room_ids,
         )
 
-        logger.info("Sending device list update notif to: %r", hosts)
-        for host in hosts:
-            self.federation_sender.send_device_messages(host)
+        if hosts:
+            logger.info("Sending device list update notif to: %r", hosts)
+            for host in hosts:
+                self.federation_sender.send_device_messages(host)
 
     @defer.inlineCallbacks
     def get_device_list_changes(self, user_id, room_ids, from_key):
+        """For a user and their joined rooms, calculate which device updates
+        we need to return.
+        """
         room_ids = frozenset(room_ids)
 
         user_ids_changed = set()
@@ -236,11 +243,14 @@ class DeviceHandler(BaseHandler):
 
         if get_domain_from_id(user_id) != origin:
             # TODO: Raise?
+            logger.warning("Got device list update edu for %r from %r", user_id, origin)
             return
 
         logger.info("Got edu: %r", edu_content)
 
         with (yield self._remote_edue_linearizer.queue(user_id)):
+            # If the prev id matches whats in our cache table, then we don't need
+            # to resync the users device list, otherwise we do.
             resync = True
             if len(prev_ids) == 1:
                 extremity = yield self.store.get_device_list_remote_extremity(user_id)
@@ -249,6 +259,7 @@ class DeviceHandler(BaseHandler):
                     resync = False
 
             if resync:
+                # Fetch all devices for the user.
                 result = yield self.federation.query_user_devices(origin, user_id)
                 stream_id = result["stream_id"]
                 devices = result["devices"]
@@ -258,6 +269,8 @@ class DeviceHandler(BaseHandler):
                 device_ids = [device["device_id"] for device in devices]
                 yield self.notify_device_update(user_id, device_ids)
             else:
+                # Simply update the single device, since we know that is the only
+                # change (becuase of the single prev_id matching the current cache)
                 content = dict(edu_content)
                 for key in ("user_id", "device_id", "stream_id", "prev_ids"):
                     content.pop(key, None)
diff --git a/synapse/handlers/e2e_keys.py b/synapse/handlers/e2e_keys.py
index 832998a6d3..a16b9def8d 100644
--- a/synapse/handlers/e2e_keys.py
+++ b/synapse/handlers/e2e_keys.py
@@ -75,7 +75,7 @@ class E2eKeysHandler(object):
             else:
                 remote_queries[user_id] = device_ids
 
-        # do the queries
+        # Firt get local devices.
         failures = {}
         results = {}
         if local_query:
@@ -84,6 +84,7 @@ class E2eKeysHandler(object):
                 if user_id in local_query:
                     results[user_id] = keys
 
+        # Now attempt to get any remote devices from our local cache.
         remote_queries_not_in_cache = {}
         if remote_queries:
             query_list = []
@@ -115,6 +116,7 @@ class E2eKeysHandler(object):
                 r = remote_queries_not_in_cache.setdefault(domain, {})
                 r[user_id] = remote_queries[user_id]
 
+        # Now fetch any devices that we don't have in our cache
         @defer.inlineCallbacks
         def do_remote_query(destination):
             destination_query = remote_queries_not_in_cache[destination]
diff --git a/synapse/storage/devices.py b/synapse/storage/devices.py
index 8ee3119db2..cf38dbaa3c 100644
--- a/synapse/storage/devices.py
+++ b/synapse/storage/devices.py
@@ -139,6 +139,9 @@ class DeviceStore(SQLBaseStore):
         defer.returnValue({d["device_id"]: d for d in devices})
 
     def get_device_list_remote_extremity(self, user_id):
+        """Get the last stream_id we got for a user. May be None if we haven't
+        got any information for them.
+        """
         return self._simple_select_one_onecol(
             table="device_lists_remote_extremeties",
             keyvalues={"user_id": user_id},
@@ -149,6 +152,8 @@ class DeviceStore(SQLBaseStore):
 
     def update_remote_device_list_cache_entry(self, user_id, device_id, content,
                                               stream_id):
+        """Updates a single user's device in the cache.
+        """
         return self.runInteraction(
             "update_remote_device_list_cache_entry",
             self._update_remote_device_list_cache_entry_txn,
@@ -181,6 +186,8 @@ class DeviceStore(SQLBaseStore):
         )
 
     def update_remote_device_list_cache(self, user_id, devices, stream_id):
+        """Replace the cache of the remote user's devices.
+        """
         return self.runInteraction(
             "update_remote_device_list_cache",
             self._update_remote_device_list_cache_txn,
@@ -222,6 +229,11 @@ class DeviceStore(SQLBaseStore):
         )
 
     def get_devices_by_remote(self, destination, from_stream_id):
+        """Get stream of updates to send to remote servers
+
+        Returns:
+            (now_stream_id, [ { updates }, .. ])
+        """
         now_stream_id = self._device_list_id_gen.get_current_token()
 
         has_changed = self._device_list_federation_stream_cache.has_entity_changed(
@@ -290,6 +302,17 @@ class DeviceStore(SQLBaseStore):
         return (now_stream_id, results)
 
     def get_user_devices_from_cache(self, query_list):
+        """Get the devices (and keys if any) for remote users from the cache.
+
+        Args:
+            query_list(list): List of (user_id, device_ids), if device_ids is
+                falsey then return all device ids for that user.
+
+        Returns:
+            (user_ids_not_in_cache, results_map), where user_ids_not_in_cache is
+            a set of user_ids and results_map is a mapping of
+            user_id -> device_id -> device_info
+        """
         return self.runInteraction(
             "get_user_devices_from_cache", self._get_user_devices_from_cache_txn,
             query_list,
@@ -347,6 +370,11 @@ class DeviceStore(SQLBaseStore):
         return user_ids_not_in_cache, results
 
     def get_devices_with_keys_by_user(self, user_id):
+        """Get all devices (with any device keys) for a user
+
+        Returns:
+            (stream_id, devices)
+        """
         return self.runInteraction(
             "get_devices_with_keys_by_user",
             self._get_devices_with_keys_by_user_txn, user_id,
@@ -380,6 +408,8 @@ class DeviceStore(SQLBaseStore):
         return now_stream_id, []
 
     def mark_as_sent_devices_by_remote(self, destination, stream_id):
+        """Mark that updates have successfully been sent to the destination.
+        """
         return self.runInteraction(
             "mark_as_sent_devices_by_remote", self._mark_as_sent_devices_by_remote_txn,
             destination, stream_id,
@@ -403,6 +433,8 @@ class DeviceStore(SQLBaseStore):
 
     @defer.inlineCallbacks
     def get_user_whose_devices_changed(self, from_key):
+        """Get set of users whose devices have changed since `from_key`.
+        """
         from_key = int(from_key)
         changed = self._device_list_stream_cache.get_all_entities_changed(from_key)
         if changed is not None:
@@ -416,8 +448,9 @@ class DeviceStore(SQLBaseStore):
 
     @defer.inlineCallbacks
     def add_device_change_to_streams(self, user_id, device_ids, hosts):
-        # device_lists_stream
-        # device_lists_outbound_pokes
+        """Persist that a user's devices have been updated, and which hosts
+        (if any) should be poked.
+        """
         with self._device_list_id_gen.get_next() as stream_id:
             yield self.runInteraction(
                 "add_device_change_to_streams", self._add_device_change_txn,
diff --git a/synapse/storage/schema/delta/40/device_list_streams.sql b/synapse/storage/schema/delta/40/device_list_streams.sql
index d1051c6ddf..8348c143c3 100644
--- a/synapse/storage/schema/delta/40/device_list_streams.sql
+++ b/synapse/storage/schema/delta/40/device_list_streams.sql
@@ -13,7 +13,7 @@
  * limitations under the License.
  */
 
-
+-- Cache of remote devices.
 CREATE TABLE device_lists_remote_cache (
     user_id TEXT NOT NULL,
     device_id TEXT NOT NULL,
@@ -23,6 +23,8 @@ CREATE TABLE device_lists_remote_cache (
 CREATE INDEX device_lists_remote_cache_id ON device_lists_remote_cache(user_id, device_id);
 
 
+-- The last update we got for a user. Empty if we're not receiving updates for
+-- that user.
 CREATE TABLE device_lists_remote_extremeties (
     user_id TEXT NOT NULL,
     stream_id TEXT NOT NULL
@@ -31,6 +33,7 @@ CREATE TABLE device_lists_remote_extremeties (
 CREATE INDEX device_lists_remote_extremeties_id ON device_lists_remote_extremeties(user_id, stream_id);
 
 
+-- Stream of device lists updates. Includes both local and remotes
 CREATE TABLE device_lists_stream (
     stream_id BIGINT NOT NULL,
     user_id TEXT NOT NULL,
@@ -40,6 +43,9 @@ CREATE TABLE device_lists_stream (
 CREATE INDEX device_lists_stream_id ON device_lists_stream(stream_id, user_id);
 
 
+-- The stream of updates to send to other servers. We keep at least one row
+-- per user that was sent so that the prev_id for any new updates can be
+-- calculated
 CREATE TABLE device_lists_outbound_pokes (
     destination TEXT NOT NULL,
     stream_id BIGINT NOT NULL,