Fix rare notifier bug where listeners dont timeout

There was a race condition that caused the notifier to 'miss' the
timeout notification, since there were no other checks for the timeout
this caused listeners to get stuck in a loop until something happened.

2 files changed, 10 insertions, 11 deletions

diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index a86996689c..b62773dcbe 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -510,6 +510,7 @@ class SyncHandler(object):
         Returns:
             Deferred(SyncResult)
         """
+        logger.info("Calculating sync response for %r", sync_config.user)
 
         # NB: The now_token gets changed by some of the generate_sync_* methods,
         # this is due to some of the underlying streams not supporting the ability
diff --git a/synapse/notifier.py b/synapse/notifier.py
index 054ca59ad2..40baa6969a 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -294,14 +294,7 @@ class Notifier(object):
 
         result = None
         if timeout:
-            # Will be set to a _NotificationListener that we'll be waiting on.
-            # Allows us to cancel it.
-            listener = None
-
-            def timed_out():
-                if listener:
-                    listener.deferred.cancel()
-            timer = self.clock.call_later(timeout / 1000., timed_out)
+            end_time = self.clock.time_msec() + timeout
 
             prev_token = from_token
             while not result:
@@ -312,6 +305,10 @@ class Notifier(object):
                     if result:
                         break
 
+                    now = self.clock.time_msec()
+                    if end_time <= now:
+                        break
+
                     # Now we wait for the _NotifierUserStream to be told there
                     # is a new token.
                     # We need to supply the token we supplied to callback so
@@ -319,11 +316,12 @@ class Notifier(object):
                     prev_token = current_token
                     listener = user_stream.new_listener(prev_token)
                     with PreserveLoggingContext():
-                        yield listener.deferred
+                        yield self.clock.time_bound_deferred(
+                            listener.deferred,
+                            time_out=(end_time - now) / 1000.
+                        )
                 except defer.CancelledError:
                     break
-
-            self.clock.cancel_call_later(timer, ignore_errs=True)
         else:
             current_token = user_stream.current_token
             result = yield callback(from_token, current_token)