Merge branch 'develop' of github.com:matrix-org/synapse into erikj/timings github/erikj/timings erikj/timings

1 files changed, 39 insertions, 34 deletions

diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 0e19f777b8..6b70fa3817 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -183,7 +183,7 @@ class PresenceHandler(object):
         # The initial delay is to allow disconnected clients a chance to
         # reconnect before we treat them as offline.
         self.clock.call_later(
-            30 * 1000,
+            30,
             self.clock.looping_call,
             self._handle_timeouts,
             5000,
@@ -283,44 +283,48 @@ class PresenceHandler(object):
         """Checks the presence of users that have timed out and updates as
         appropriate.
         """
+        logger.info("Handling presence timeouts")
         now = self.clock.time_msec()
 
-        with Measure(self.clock, "presence_handle_timeouts"):
-            # Fetch the list of users that *may* have timed out. Things may have
-            # changed since the timeout was set, so we won't necessarily have to
-            # take any action.
-            users_to_check = set(self.wheel_timer.fetch(now))
-
-            # Check whether the lists of syncing processes from an external
-            # process have expired.
-            expired_process_ids = [
-                process_id for process_id, last_update
-                in self.external_process_last_update.items()
-                if now - last_update > EXTERNAL_PROCESS_EXPIRY
-            ]
-            for process_id in expired_process_ids:
-                users_to_check.update(
-                    self.external_process_to_current_syncs.pop(process_id, ())
-                )
-                self.external_process_last_update.pop(process_id)
+        try:
+            with Measure(self.clock, "presence_handle_timeouts"):
+                # Fetch the list of users that *may* have timed out. Things may have
+                # changed since the timeout was set, so we won't necessarily have to
+                # take any action.
+                users_to_check = set(self.wheel_timer.fetch(now))
+
+                # Check whether the lists of syncing processes from an external
+                # process have expired.
+                expired_process_ids = [
+                    process_id for process_id, last_update
+                    in self.external_process_last_updated_ms.items()
+                    if now - last_update > EXTERNAL_PROCESS_EXPIRY
+                ]
+                for process_id in expired_process_ids:
+                    users_to_check.update(
+                        self.external_process_last_updated_ms.pop(process_id, ())
+                    )
+                    self.external_process_last_update.pop(process_id)
 
-            states = [
-                self.user_to_current_state.get(
-                    user_id, UserPresenceState.default(user_id)
-                )
-                for user_id in users_to_check
-            ]
+                states = [
+                    self.user_to_current_state.get(
+                        user_id, UserPresenceState.default(user_id)
+                    )
+                    for user_id in users_to_check
+                ]
 
-            timers_fired_counter.inc_by(len(states))
+                timers_fired_counter.inc_by(len(states))
 
-            changes = handle_timeouts(
-                states,
-                is_mine_fn=self.is_mine_id,
-                syncing_users=self.get_syncing_users(),
-                now=now,
-            )
+                changes = handle_timeouts(
+                    states,
+                    is_mine_fn=self.is_mine_id,
+                    syncing_user_ids=self.get_currently_syncing_users(),
+                    now=now,
+                )
 
-        preserve_fn(self._update_states)(changes)
+            preserve_fn(self._update_states)(changes)
+        except:
+            logger.exception("Exception in _handle_timeouts loop")
 
     @defer.inlineCallbacks
     def bump_presence_active_time(self, user):
@@ -402,7 +406,8 @@ class PresenceHandler(object):
             user_id for user_id, count in self.user_to_num_current_syncs.items()
             if count
         }
-        syncing_user_ids.update(self.external_process_to_current_syncs.values())
+        for user_ids in self.external_process_to_current_syncs.values():
+            syncing_user_ids.update(user_ids)
         return syncing_user_ids
 
     @defer.inlineCallbacks