From 5c75adff951b27744528d9c095f4ff1f8df1f77a Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Tue, 12 May 2015 11:00:37 +0100
Subject: [PATCH 01/11] Add a NotifierUserStream to hold all the notification
 listeners for a user

---
 synapse/notifier.py | 228 ++++++++++++++++++++++----------------------
 1 file changed, 115 insertions(+), 113 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index abe12b143..0b50898f3 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -43,28 +43,18 @@ def count(func, l):
 
 class _NotificationListener(object):
     """ This represents a single client connection to the events stream.
-
     The events stream handler will have yielded to the deferred, so to
     notify the handler it is sufficient to resolve the deferred.
-
-    This listener will also keep track of which rooms it is listening in
-    so that it can remove itself from the indexes in the Notifier class.
     """
 
-    def __init__(self, user, rooms, deferred, appservice=None):
-        self.user = user
-        self.appservice = appservice
+    def __init__(self, deferred):
         self.deferred = deferred
-        self.rooms = rooms
-        self.timer = None
 
     def notified(self):
         return self.deferred.called
 
-    def notify(self, notifier):
-        """ Inform whoever is listening about the new events. This will
-        also remove this listener from all the indexes in the Notifier
-        it knows about.
+    def notify(self):
+        """ Inform whoever is listening about the new events.
         """
 
         try:
@@ -72,27 +62,45 @@ class _NotificationListener(object):
         except defer.AlreadyCalledError:
             pass
 
-        # Should the following be done be using intrusively linked lists?
-        # -- erikj
+
+class _NotifierUserStream(object):
+    """This represents a user connected to the event stream.
+    It tracks the most recent stream token for that user.
+    At a given point a user may have a number of streams listening for
+    events.
+
+    This listener will also keep track of which rooms it is listening in
+    so that it can remove itself from the indexes in the Notifier class.
+    """
+
+    def __init__(self, user, rooms, current_token, appservice=None):
+        self.user = user
+        self.appservice = appservice
+        self.listeners = set()
+        self.rooms = rooms
+        self.current_token = current_token
+
+    def notify(self, new_token):
+        for listener in self.listeners:
+            listener.notify(new_token)
+        self.listeners.clear()
+
+    def remove(self, notifier):
+        """ Remove this listener from all the indexes in the Notifier
+        it knows about.
+        """
 
         for room in self.rooms:
-            lst = notifier.room_to_listeners.get(room, set())
+            lst = notifier.room_to_user_streams.get(room, set())
             lst.discard(self)
 
-        notifier.user_to_listeners.get(self.user, set()).discard(self)
+        notifier.user_to_user_streams.get(self.user, set()).discard(self)
 
         if self.appservice:
-            notifier.appservice_to_listeners.get(
+            notifier.appservice_to_user_streams.get(
                 self.appservice, set()
             ).discard(self)
 
-        # Cancel the timeout for this notifer if one exists.
-        if self.timer is not None:
-            try:
-                notifier.clock.cancel_call_later(self.timer)
-            except:
-                logger.warn("Failed to cancel notifier timer")
-
 
 class Notifier(object):
     """ This class is responsible for notifying any listeners when there are
@@ -104,11 +112,12 @@ class Notifier(object):
     def __init__(self, hs):
         self.hs = hs
 
-        self.room_to_listeners = {}
-        self.user_to_listeners = {}
-        self.appservice_to_listeners = {}
+        self.user_to_user_stream = {}
+        self.room_to_user_streams = {}
+        self.appservice_to_user_streams = {}
 
         self.event_sources = hs.get_event_sources()
+        self.store = hs.get_datastore()
 
         self.clock = hs.get_clock()
 
@@ -120,34 +129,34 @@ class Notifier(object):
         # when rendering the metrics page, which is likely once per minute at
         # most when scraping it.
         def count_listeners():
-            all_listeners = set()
+            all_user_streams = set()
 
-            for x in self.room_to_listeners.values():
-                all_listeners |= x
-            for x in self.user_to_listeners.values():
-                all_listeners |= x
-            for x in self.appservice_to_listeners.values():
-                all_listeners |= x
+            for x in self.room_to_user_streams.values():
+                all_user_streams |= x
+            for x in self.user_to_user_streams.values():
+                all_user_streams |= x
+            for x in self.appservice_to_user_streams.values():
+                all_user_streams |= x
 
-            return len(all_listeners)
+            return sum(len(stream.listeners) for stream in all_user_streams)
         metrics.register_callback("listeners", count_listeners)
 
         metrics.register_callback(
             "rooms",
-            lambda: count(bool, self.room_to_listeners.values()),
+            lambda: count(bool, self.room_to_user_streams.values()),
         )
         metrics.register_callback(
             "users",
-            lambda: count(bool, self.user_to_listeners.values()),
+            lambda: len(self.user_to_user_stream),
         )
         metrics.register_callback(
             "appservices",
-            lambda: count(bool, self.appservice_to_listeners.values()),
+            lambda: count(bool, self.appservice_to_user_streams.values()),
         )
 
     @log_function
     @defer.inlineCallbacks
-    def on_new_room_event(self, event, extra_users=[]):
+    def on_new_room_event(self, event, new_token, extra_users=[]):
         """ Used by handlers to inform the notifier something has happened
         in the room, room event wise.
 
@@ -155,6 +164,7 @@ class Notifier(object):
         listening to the room, and any listeners for the users in the
         `extra_users` param.
         """
+        assert isinstance(new_token, StreamToken)
         yield run_on_reactor()
         # poke any interested application service.
         self.hs.get_handlers().appservice_handler.notify_interested_services(
@@ -163,72 +173,60 @@ class Notifier(object):
 
         room_id = event.room_id
 
-        room_listeners = self.room_to_listeners.get(room_id, set())
+        room_user_streams = self.room_to_user_streams.get(room_id, set())
 
-        _discard_if_notified(room_listeners)
-
-        listeners = room_listeners.copy()
+        user_streams = room_user_streams.copy()
 
         for user in extra_users:
-            user_listeners = self.user_to_listeners.get(user, set())
+            user_stream = self.user_to_user_stream.get(user)
+            if user_stream is not None:
+                user_streams.add(user_stream)
 
-            _discard_if_notified(user_listeners)
-
-            listeners |= user_listeners
-
-        for appservice in self.appservice_to_listeners:
+        for appservice in self.appservice_to_user_streams:
             # TODO (kegan): Redundant appservice listener checks?
-            # App services will already be in the room_to_listeners set, but
+            # App services will already be in the room_to_user_streams set, but
             # that isn't enough. They need to be checked here in order to
             # receive *invites* for users they are interested in. Does this
-            # make the room_to_listeners check somewhat obselete?
+            # make the room_to_user_streams check somewhat obselete?
             if appservice.is_interested(event):
-                app_listeners = self.appservice_to_listeners.get(
+                app_user_streams = self.appservice_to_user_streams.get(
                     appservice, set()
                 )
+                user_streams |= app_user_streams
 
-                _discard_if_notified(app_listeners)
-
-                listeners |= app_listeners
-
-        logger.debug("on_new_room_event listeners %s", listeners)
+        logger.debug("on_new_room_event listeners %s", user_streams)
 
         with PreserveLoggingContext():
-            for listener in listeners:
+            for user_stream in user_streams:
                 try:
-                    listener.notify(self)
+                    user_stream.notify(new_token)
                 except:
                     logger.exception("Failed to notify listener")
 
     @defer.inlineCallbacks
     @log_function
-    def on_new_user_event(self, users=[], rooms=[]):
+    def on_new_user_event(self, new_token, users=[], rooms=[]):
         """ Used to inform listeners that something has happend
         presence/user event wise.
 
         Will wake up all listeners for the given users and rooms.
         """
+        assert isinstance(new_token, StreamToken)
         yield run_on_reactor()
-        listeners = set()
+        user_streams = set()
 
         for user in users:
-            user_listeners = self.user_to_listeners.get(user, set())
-
-            _discard_if_notified(user_listeners)
-
-            listeners |= user_listeners
+            user_stream = self.user_to_user_stream.get(user)
+            if user_stream:
+                user_stream.add(user_stream)
 
         for room in rooms:
-            room_listeners = self.room_to_listeners.get(room, set())
-
-            _discard_if_notified(room_listeners)
-
-            listeners |= room_listeners
+            user_streams |= self.room_to_user_streams.get(room, set())
 
         with PreserveLoggingContext():
-            for listener in listeners:
+            for user_stream in user_streams:
                 try:
-                    listener.notify(self)
+                    user_streams.notify(new_token)
                 except:
                     logger.exception("Failed to notify listener")
 
@@ -240,21 +238,32 @@ class Notifier(object):
         """
 
         deferred = defer.Deferred()
-        appservice = yield self.hs.get_datastore().get_app_service_by_user_id(
-            user.to_string()
-        )
 
-        listener = [_NotificationListener(
-            user=user,
-            rooms=rooms,
-            deferred=deferred,
-            appservice=appservice,
-        )]
+        user_stream = self.user_to_user_streams.get(user)
+        if user_stream is None:
+            appservice = yield self.store.get_app_service_by_user_id(
+                user.to_string()
+            )
+            current_token = yield self.event_sources.get_current_token()
+            user_stream = _NotifierUserStream(
+                user=user,
+                rooms=rooms,
+                appservice=appservice,
+                current_token=current_token,
+            )
+            self._register_with_keys(user_stream)
+        else:
+            current_token = user_stream.current_token
 
-        if timeout:
-            self._register_with_keys(listener[0])
+        if timeout and not current_token.is_after(from_token):
+            listener = [_NotificationListener(deferred)]
+            user_stream.listeners.add(listener[0])
+
+        if current_token.is_after(from_token):
+            result = yield callback(from_token, current_token)
+        else:
+            result = None
 
-        result = yield callback()
         timer = [None]
 
         if timeout:
@@ -263,23 +272,19 @@ class Notifier(object):
             def _timeout_listener():
                 timed_out[0] = True
                 timer[0] = None
-                listener[0].notify(self)
+                listener[0].notify(user_stream)
 
             # We create multiple notification listeners so we have to manage
             # canceling the timeout ourselves.
             timer[0] = self.clock.call_later(timeout/1000., _timeout_listener)
 
             while not result and not timed_out[0]:
-                yield deferred
+                new_token = yield deferred
                 deferred = defer.Deferred()
-                listener[0] = _NotificationListener(
-                    user=user,
-                    rooms=rooms,
-                    deferred=deferred,
-                    appservice=appservice,
-                )
-                self._register_with_keys(listener[0])
-                result = yield callback()
+                listener[0] = _NotificationListener(deferred)
+                user_stream.listeners.add(listener[0])
+                result = yield callback(current_token, new_token)
+                current_token = new_token
 
         if timer[0] is not None:
             try:
@@ -302,7 +307,7 @@ class Notifier(object):
         limit = pagination_config.limit
 
         @defer.inlineCallbacks
-        def check_for_updates():
+        def check_for_updates(start_token, end_token):
             events = []
             end_token = from_token
             for name, source in self.event_sources.sources.items():
@@ -328,26 +333,23 @@ class Notifier(object):
         defer.returnValue(result)
 
     @log_function
-    def _register_with_keys(self, listener):
-        for room in listener.rooms:
-            s = self.room_to_listeners.setdefault(room, set())
-            s.add(listener)
+    def _register_with_keys(self, user_stream):
+        self.user_to_user_stream[user_stream.user] = user_stream
 
-        self.user_to_listeners.setdefault(listener.user, set()).add(listener)
+        for room in user_stream.rooms:
+            s = self.room_to_user_stream.setdefault(room, set())
+            s.add(user_stream)
 
-        if listener.appservice:
-            self.appservice_to_listeners.setdefault(
-                listener.appservice, set()
-            ).add(listener)
+        if user_stream.appservice:
+            self.appservice_to_user_stream.setdefault(
+                user_stream.appservice, set()
+            ).add(user_stream)
 
     def _user_joined_room(self, user, room_id):
-        new_listeners = self.user_to_listeners.get(user, set())
-
-        listeners = self.room_to_listeners.setdefault(room_id, set())
-        listeners |= new_listeners
-
-        for l in new_listeners:
-            l.rooms.add(room_id)
+        new_user_stream = self.user_to_user_stream.get(user)
+        room_streams = self.room_to_user_streams.setdefault(room_id, set())
+        room_streams.add(new_user_stream)
+        new_user_stream.rooms.add(room_id)
 
 
 def _discard_if_notified(listener_set):

From 63878c03794d33a8767425e114845159e5c1cb9a Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Wed, 13 May 2015 13:42:21 +0100
Subject: [PATCH 02/11] Don't bother checking for updates if the stream token
 hasn't advanced for a user

---
 synapse/handlers/_base.py             |  7 ++-
 synapse/handlers/federation.py        | 25 +++++----
 synapse/handlers/presence.py          |  4 ++
 synapse/handlers/typing.py            |  4 +-
 synapse/notifier.py                   | 75 ++++++++++++++++++---------
 synapse/storage/events.py             |  3 ++
 synapse/types.py                      | 19 ++++++-
 tests/handlers/test_federation.py     |  4 +-
 tests/handlers/test_room.py           |  8 +--
 tests/handlers/test_typing.py         | 12 ++---
 tests/rest/client/v1/test_presence.py | 15 ++++--
 tests/utils.py                        |  2 +-
 12 files changed, 123 insertions(+), 55 deletions(-)

diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index ddc5c21e7..833ff4137 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -105,7 +105,9 @@ class BaseHandler(object):
         if not suppress_auth:
             self.auth.check(event, auth_events=context.current_state)
 
-        yield self.store.persist_event(event, context=context)
+        (event_stream_id, max_stream_id) = yield self.store.persist_event(
+            event, context=context
+        )
 
         federation_handler = self.hs.get_handlers().federation_handler
 
@@ -142,7 +144,8 @@ class BaseHandler(object):
         with PreserveLoggingContext():
             # Don't block waiting on waking up all the listeners.
             notify_d = self.notifier.on_new_room_event(
-                event, extra_users=extra_users
+                event, event_stream_id, max_stream_id,
+                extra_users=extra_users
             )
 
         def log_failure(f):
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 7d9906039..bc0f7b0ee 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -160,7 +160,7 @@ class FederationHandler(BaseHandler):
                     )
 
         try:
-            yield self._handle_new_event(
+            _, event_stream_id, max_stream_id = yield self._handle_new_event(
                 origin,
                 event,
                 state=state,
@@ -203,7 +203,8 @@ class FederationHandler(BaseHandler):
 
             with PreserveLoggingContext():
                 d = self.notifier.on_new_room_event(
-                    event, extra_users=extra_users
+                    event, event_stream_id, max_stream_id,
+                    extra_users=extra_users
                 )
 
             def log_failure(f):
@@ -561,7 +562,7 @@ class FederationHandler(BaseHandler):
                 if e.event_id in auth_ids
             }
 
-            yield self._handle_new_event(
+            _, event_stream_id, max_stream_id = yield self._handle_new_event(
                 origin,
                 new_event,
                 state=state,
@@ -571,7 +572,8 @@ class FederationHandler(BaseHandler):
 
             with PreserveLoggingContext():
                 d = self.notifier.on_new_room_event(
-                    new_event, extra_users=[joinee]
+                    new_event, event_stream_id, max_stream_id,
+                    extra_users=[joinee]
                 )
 
             def log_failure(f):
@@ -637,7 +639,9 @@ class FederationHandler(BaseHandler):
 
         event.internal_metadata.outlier = False
 
-        context = yield self._handle_new_event(origin, event)
+        context, event_stream_id, max_stream_id = yield self._handle_new_event(
+            origin, event
+        )
 
         logger.debug(
             "on_send_join_request: After _handle_new_event: %s, sigs: %s",
@@ -653,7 +657,7 @@ class FederationHandler(BaseHandler):
 
         with PreserveLoggingContext():
             d = self.notifier.on_new_room_event(
-                event, extra_users=extra_users
+                event, event_stream_id, max_stream_id, extra_users=extra_users
             )
 
         def log_failure(f):
@@ -727,7 +731,7 @@ class FederationHandler(BaseHandler):
 
         context = yield self.state_handler.compute_event_context(event)
 
-        yield self.store.persist_event(
+        event_stream_id, max_stream_id = yield self.store.persist_event(
             event,
             context=context,
             backfilled=False,
@@ -736,7 +740,8 @@ class FederationHandler(BaseHandler):
         target_user = UserID.from_string(event.state_key)
         with PreserveLoggingContext():
             d = self.notifier.on_new_room_event(
-                event, extra_users=[target_user],
+                event, event_stream_id, max_stream_id,
+                extra_users=[target_user],
             )
 
         def log_failure(f):
@@ -914,7 +919,7 @@ class FederationHandler(BaseHandler):
             )
             raise
 
-        yield self.store.persist_event(
+        event_stream_id, max_stream_id = yield self.store.persist_event(
             event,
             context=context,
             backfilled=backfilled,
@@ -922,7 +927,7 @@ class FederationHandler(BaseHandler):
             current_state=current_state,
         )
 
-        defer.returnValue(context)
+        defer.returnValue((context, event_stream_id, max_stream_id))
 
     @defer.inlineCallbacks
     def on_query_auth(self, origin, event_id, remote_auth_chain, rejects,
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 28688d532..7db4b062d 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -345,6 +345,8 @@ class PresenceHandler(BaseHandler):
         curr_users = yield rm_handler.get_room_members(room_id)
 
         for local_user in [c for c in curr_users if self.hs.is_mine(c)]:
+            statuscache = self._get_or_offline_usercache(local_user)
+            statuscache.update({}, serial=self._user_cachemap_latest_serial)
             self.push_update_to_local_and_remote(
                 observed_user=local_user,
                 users_to_push=[user],
@@ -820,6 +822,8 @@ class PresenceHandler(BaseHandler):
                                room_ids=[], statuscache=None):
         with PreserveLoggingContext():
             self.notifier.on_new_user_event(
+                "presence_key",
+                self._user_cachemap_latest_serial,
                 users_to_push,
                 room_ids,
             )
diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index 64fe51aa3..a9895292c 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -218,7 +218,9 @@ class TypingNotificationHandler(BaseHandler):
         self._room_serials[room_id] = self._latest_room_serial
 
         with PreserveLoggingContext():
-            self.notifier.on_new_user_event(rooms=[room_id])
+            self.notifier.on_new_user_event(
+                "typing_key", self._latest_room_serial, rooms=[room_id]
+            )
 
 
 class TypingNotificationEventSource(object):
diff --git a/synapse/notifier.py b/synapse/notifier.py
index 214a2b28c..4d10c0503 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -52,12 +52,11 @@ class _NotificationListener(object):
     def notified(self):
         return self.deferred.called
 
-    def notify(self):
+    def notify(self, token):
         """ Inform whoever is listening about the new events.
         """
-
         try:
-            self.deferred.callback(None)
+            self.deferred.callback(token)
         except defer.AlreadyCalledError:
             pass
 
@@ -73,15 +72,18 @@ class _NotifierUserStream(object):
     """
 
     def __init__(self, user, rooms, current_token, appservice=None):
-        self.user = user
+        self.user = str(user)
         self.appservice = appservice
         self.listeners = set()
-        self.rooms = rooms
+        self.rooms = set(rooms)
         self.current_token = current_token
 
-    def notify(self, new_token):
+    def notify(self, stream_key, stream_id):
+        self.current_token = self.current_token.copy_and_replace(
+            stream_key, stream_id
+        )
         for listener in self.listeners:
-            listener.notify(new_token)
+            listener.notify(self.current_token)
         self.listeners.clear()
 
     def remove(self, notifier):
@@ -117,6 +119,7 @@ class Notifier(object):
 
         self.event_sources = hs.get_event_sources()
         self.store = hs.get_datastore()
+        self.pending_new_room_events = []
 
         self.clock = hs.get_clock()
 
@@ -153,9 +156,21 @@ class Notifier(object):
             lambda: count(bool, self.appservice_to_user_streams.values()),
         )
 
+    def notify_pending_new_room_events(self, max_room_stream_id):
+        pending = sorted(self.pending_new_room_events)
+        self.pending_new_room_events = []
+        for event, room_stream_id, extra_users in pending:
+            if room_stream_id > max_room_stream_id:
+                self.pending_new_room_events.append((
+                    event, room_stream_id, extra_users
+                ))
+            else:
+                self._on_new_room_event(event, room_stream_id, extra_users)
+
     @log_function
     @defer.inlineCallbacks
-    def on_new_room_event(self, event, new_token, extra_users=[]):
+    def on_new_room_event(self, event, room_stream_id, max_room_stream_id,
+                          extra_users=[]):
         """ Used by handlers to inform the notifier something has happened
         in the room, room event wise.
 
@@ -163,8 +178,18 @@ class Notifier(object):
         listening to the room, and any listeners for the users in the
         `extra_users` param.
         """
-        assert isinstance(new_token, StreamToken)
         yield run_on_reactor()
+
+        self.notify_pending_new_room_events(max_room_stream_id)
+
+        if room_stream_id > max_room_stream_id:
+            self.pending_new_room_events.append((
+                event, room_stream_id, extra_users
+            ))
+        else:
+            self._on_new_room_event(event, room_stream_id, extra_users)
+
+    def _on_new_room_event(self, event, room_stream_id, extra_users=[]):
         # poke any interested application service.
         self.hs.get_handlers().appservice_handler.notify_interested_services(
             event
@@ -197,33 +222,32 @@ class Notifier(object):
 
         for user_stream in user_streams:
             try:
-                user_stream.notify(new_token)
+                user_stream.notify("room_key", "s%d" % (room_stream_id,))
             except:
                 logger.exception("Failed to notify listener")
 
     @defer.inlineCallbacks
     @log_function
-    def on_new_user_event(self, new_token, users=[], rooms=[]):
+    def on_new_user_event(self, stream_key, new_token, users=[], rooms=[]):
         """ Used to inform listeners that something has happend
         presence/user event wise.
 
         Will wake up all listeners for the given users and rooms.
         """
-        assert isinstance(new_token, StreamToken)
         yield run_on_reactor()
         user_streams = set()
 
         for user in users:
             user_stream = self.user_to_user_stream.get(user)
-            if user_stream:
-                user_stream.add(user_stream)
+            if user_stream is not None:
+                user_streams.add(user_stream)
 
         for room in rooms:
             user_streams |= self.room_to_user_streams.get(room, set())
 
         for user_stream in user_streams:
             try:
-                user_streams.notify(new_token)
+                user_stream.notify(stream_key, new_token)
             except:
                 logger.exception("Failed to notify listener")
 
@@ -236,12 +260,12 @@ class Notifier(object):
 
         deferred = defer.Deferred()
 
-        user_stream = self.user_to_user_streams.get(user)
+        user = str(user)
+        user_stream = self.user_to_user_stream.get(user)
         if user_stream is None:
-            appservice = yield self.store.get_app_service_by_user_id(
-                user.to_string()
-            )
+            appservice = yield self.store.get_app_service_by_user_id(user)
             current_token = yield self.event_sources.get_current_token()
+            rooms = yield self.store.get_rooms_for_user(user)
             user_stream = _NotifierUserStream(
                 user=user,
                 rooms=rooms,
@@ -252,8 +276,9 @@ class Notifier(object):
         else:
             current_token = user_stream.current_token
 
+        listener = [_NotificationListener(deferred)]
+
         if timeout and not current_token.is_after(from_token):
-            listener = [_NotificationListener(deferred)]
             user_stream.listeners.add(listener[0])
 
         if current_token.is_after(from_token):
@@ -334,7 +359,7 @@ class Notifier(object):
         self.user_to_user_stream[user_stream.user] = user_stream
 
         for room in user_stream.rooms:
-            s = self.room_to_user_stream.setdefault(room, set())
+            s = self.room_to_user_streams.setdefault(room, set())
             s.add(user_stream)
 
         if user_stream.appservice:
@@ -343,10 +368,12 @@ class Notifier(object):
             ).add(user_stream)
 
     def _user_joined_room(self, user, room_id):
+        user = str(user)
         new_user_stream = self.user_to_user_stream.get(user)
-        room_streams = self.room_to_user_streams.setdefault(room_id, set())
-        room_streams.add(new_user_stream)
-        new_user_stream.rooms.add(room_id)
+        if new_user_stream is not None:
+            room_streams = self.room_to_user_streams.setdefault(room_id, set())
+            room_streams.add(new_user_stream)
+            new_user_stream.rooms.add(room_id)
 
 
 def _discard_if_notified(listener_set):
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index a5a686907..7d6df5f4c 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -64,6 +64,9 @@ class EventsStore(SQLBaseStore):
         except _RollbackButIsFineException:
             pass
 
+        max_persisted_id = yield self._stream_id_gen.get_max_token(self)
+        defer.returnValue((stream_ordering, max_persisted_id))
+
     @defer.inlineCallbacks
     def get_event(self, event_id, check_redacted=True,
                   get_prev_content=False, allow_rejected=False,
diff --git a/synapse/types.py b/synapse/types.py
index 0f16867d7..d89a04f7c 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -70,6 +70,8 @@ class DomainSpecificString(
         """Return a string encoding the fields of the structure object."""
         return "%s%s:%s" % (self.SIGIL, self.localpart, self.domain)
 
+    __str__ = to_string
+
     @classmethod
     def create(cls, localpart, domain,):
         return cls(localpart=localpart, domain=domain)
@@ -107,7 +109,6 @@ class StreamToken(
     def from_string(cls, string):
         try:
             keys = string.split(cls._SEPARATOR)
-
             return cls(*keys)
         except:
             raise SynapseError(400, "Invalid Token")
@@ -115,6 +116,22 @@ class StreamToken(
     def to_string(self):
         return self._SEPARATOR.join([str(k) for k in self])
 
+    @property
+    def room_stream_id(self):
+        # TODO(markjh): Awful hack to work around hacks in the presence tests
+        if type(self.room_key) is int:
+            return self.room_key
+        else:
+            return int(self.room_key[1:].split("-")[-1])
+
+    def is_after(self, other_token):
+        """Does this token contain events that the other doesn't?"""
+        return (
+            (other_token.room_stream_id < self.room_stream_id)
+            or (int(other_token.presence_key) < int(self.presence_key))
+            or (int(other_token.typing_key) < int(self.typing_key))
+        )
+
     def copy_and_replace(self, key, new_value):
         d = self._asdict()
         d[key] = new_value
diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py
index 08d2404b6..f3821242b 100644
--- a/tests/handlers/test_federation.py
+++ b/tests/handlers/test_federation.py
@@ -83,7 +83,7 @@ class FederationTestCase(unittest.TestCase):
             "hashes": {"sha256":"AcLrgtUIqqwaGoHhrEvYG1YLDIsVPYJdSRGhkp3jJp8"},
         })
 
-        self.datastore.persist_event.return_value = defer.succeed(None)
+        self.datastore.persist_event.return_value = defer.succeed((1,1))
         self.datastore.get_room.return_value = defer.succeed(True)
         self.auth.check_host_in_room.return_value = defer.succeed(True)
 
@@ -126,5 +126,5 @@ class FederationTestCase(unittest.TestCase):
         self.auth.check.assert_called_once_with(ANY, auth_events={})
 
         self.notifier.on_new_room_event.assert_called_once_with(
-            ANY, extra_users=[]
+            ANY, 1, 1, extra_users=[]
         )
diff --git a/tests/handlers/test_room.py b/tests/handlers/test_room.py
index 6417f7330..a2d763599 100644
--- a/tests/handlers/test_room.py
+++ b/tests/handlers/test_room.py
@@ -87,6 +87,8 @@ class RoomMemberHandlerTestCase(unittest.TestCase):
         self.ratelimiter = hs.get_ratelimiter()
         self.ratelimiter.send_message.return_value = (True, 0)
 
+        self.datastore.persist_event.return_value = (1,1)
+
     @defer.inlineCallbacks
     def test_invite(self):
         room_id = "!foo:red"
@@ -160,7 +162,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase):
             event, context=context,
         )
         self.notifier.on_new_room_event.assert_called_once_with(
-            event, extra_users=[UserID.from_string(target_user_id)]
+            event, 1, 1, extra_users=[UserID.from_string(target_user_id)]
         )
         self.assertFalse(self.datastore.get_room.called)
         self.assertFalse(self.datastore.store_room.called)
@@ -226,7 +228,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase):
             event, context=context
         )
         self.notifier.on_new_room_event.assert_called_once_with(
-            event, extra_users=[user]
+            event, 1, 1, extra_users=[user]
         )
 
         join_signal_observer.assert_called_with(
@@ -304,7 +306,7 @@ class RoomMemberHandlerTestCase(unittest.TestCase):
             event, context=context
         )
         self.notifier.on_new_room_event.assert_called_once_with(
-            event, extra_users=[user]
+            event, 1, 1, extra_users=[user]
         )
 
         leave_signal_observer.assert_called_with(
diff --git a/tests/handlers/test_typing.py b/tests/handlers/test_typing.py
index b318d4944..7ccbe2ea9 100644
--- a/tests/handlers/test_typing.py
+++ b/tests/handlers/test_typing.py
@@ -183,7 +183,7 @@ class TypingNotificationsTestCase(unittest.TestCase):
         )
 
         self.on_new_user_event.assert_has_calls([
-            call(rooms=[self.room_id]),
+            call('typing_key', 1, rooms=[self.room_id]),
         ])
 
         self.assertEquals(self.event_source.get_current_key(), 1)
@@ -246,7 +246,7 @@ class TypingNotificationsTestCase(unittest.TestCase):
         )
 
         self.on_new_user_event.assert_has_calls([
-            call(rooms=[self.room_id]),
+            call('typing_key', 1, rooms=[self.room_id]),
         ])
 
         self.assertEquals(self.event_source.get_current_key(), 1)
@@ -300,7 +300,7 @@ class TypingNotificationsTestCase(unittest.TestCase):
         )
 
         self.on_new_user_event.assert_has_calls([
-            call(rooms=[self.room_id]),
+            call('typing_key', 1, rooms=[self.room_id]),
         ])
 
         yield put_json.await_calls()
@@ -332,7 +332,7 @@ class TypingNotificationsTestCase(unittest.TestCase):
         )
 
         self.on_new_user_event.assert_has_calls([
-            call(rooms=[self.room_id]),
+            call('typing_key', 1, rooms=[self.room_id]),
         ])
         self.on_new_user_event.reset_mock()
 
@@ -352,7 +352,7 @@ class TypingNotificationsTestCase(unittest.TestCase):
         self.clock.advance_time(11)
 
         self.on_new_user_event.assert_has_calls([
-            call(rooms=[self.room_id]),
+            call('typing_key', 2, rooms=[self.room_id]),
         ])
 
         self.assertEquals(self.event_source.get_current_key(), 2)
@@ -378,7 +378,7 @@ class TypingNotificationsTestCase(unittest.TestCase):
         )
 
         self.on_new_user_event.assert_has_calls([
-            call(rooms=[self.room_id]),
+            call('typing_key', 3, rooms=[self.room_id]),
         ])
         self.on_new_user_event.reset_mock()
 
diff --git a/tests/rest/client/v1/test_presence.py b/tests/rest/client/v1/test_presence.py
index 8e0c5fa63..c0c52796a 100644
--- a/tests/rest/client/v1/test_presence.py
+++ b/tests/rest/client/v1/test_presence.py
@@ -27,6 +27,7 @@ from synapse.handlers.presence import PresenceHandler
 from synapse.rest.client.v1 import presence
 from synapse.rest.client.v1 import events
 from synapse.types import UserID
+from synapse.util.async import run_on_reactor
 
 
 OFFLINE = PresenceState.OFFLINE
@@ -264,6 +265,7 @@ class PresenceEventStreamTestCase(unittest.TestCase):
             datastore=Mock(spec=[
                 "set_presence_state",
                 "get_presence_list",
+                "get_rooms_for_user",
             ]),
             clock=Mock(spec=[
                 "call_later",
@@ -298,6 +300,9 @@ class PresenceEventStreamTestCase(unittest.TestCase):
         self.mock_datastore.get_app_service_by_user_id = Mock(
             return_value=defer.succeed(None)
         )
+        self.mock_datastore.get_rooms_for_user = (
+            lambda u: get_rooms_for_user(UserID.from_string(u))
+        )
 
         def get_profile_displayname(user_id):
             return defer.succeed("Frank")
@@ -350,19 +355,19 @@ class PresenceEventStreamTestCase(unittest.TestCase):
         self.mock_datastore.set_presence_state.return_value = defer.succeed(
             {"state": ONLINE}
         )
-        self.mock_datastore.get_presence_list.return_value = defer.succeed(
-            []
-        )
+        self.mock_datastore.get_presence_list.return_value = defer.succeed([])
 
         yield self.presence.set_state(self.u_banana, self.u_banana,
             state={"presence": ONLINE}
         )
 
+        yield run_on_reactor()
+
         (code, response) = yield self.mock_resource.trigger("GET",
-                "/events?from=0_1_0&timeout=0", None)
+                "/events?from=s0_1_0&timeout=0", None)
 
         self.assertEquals(200, code)
-        self.assertEquals({"start": "0_1_0", "end": "0_2_0", "chunk": [
+        self.assertEquals({"start": "s0_1_0", "end": "s0_2_0", "chunk": [
             {"type": "m.presence",
              "content": {
                  "user_id": "@banana:test",
diff --git a/tests/utils.py b/tests/utils.py
index cc038fecf..a67530bd6 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -355,7 +355,7 @@ class MemoryDataStore(object):
         return []
 
     def get_room_events_max_id(self):
-        return 0  # TODO (erikj)
+        return "s0"  # TODO (erikj)
 
     def get_send_event_level(self, room_id):
         return defer.succeed(0)

From df6db5c8025e67c410a60994ac87eb39d842af30 Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Wed, 13 May 2015 15:08:24 +0100
Subject: [PATCH 03/11] Don't bother checking for new events from a source if
 the stream token hasn't advanced for that source

---
 synapse/notifier.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index 4d10c0503..d2fefea75 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -294,7 +294,7 @@ class Notifier(object):
             def _timeout_listener():
                 timed_out[0] = True
                 timer[0] = None
-                listener[0].notify(user_stream)
+                listener[0].notify(from_token)
 
             # We create multiple notification listeners so we have to manage
             # canceling the timeout ourselves.
@@ -329,11 +329,15 @@ class Notifier(object):
         limit = pagination_config.limit
 
         @defer.inlineCallbacks
-        def check_for_updates(start_token, end_token):
+        def check_for_updates(before_token, after_token):
             events = []
             end_token = from_token
             for name, source in self.event_sources.sources.items():
                 keyname = "%s_key" % name
+                before_id = getattr(before_token, keyname)
+                after_id = getattr(after_token, keyname)
+                if before_id == after_id:
+                    continue
                 stuff, new_key = yield source.get_new_events_for_user(
                     user, getattr(from_token, keyname), limit,
                 )

From 9af432257da39385ce03c846b3c76f7fc7778ff0 Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Wed, 13 May 2015 15:42:13 +0100
Subject: [PATCH 04/11] Don't set a timer if there's already a result to return

---
 synapse/notifier.py | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index d2fefea75..6fcb7767a 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -82,9 +82,10 @@ class _NotifierUserStream(object):
         self.current_token = self.current_token.copy_and_replace(
             stream_key, stream_id
         )
-        for listener in self.listeners:
+        listeners = self.listeners
+        self.listeners = set()
+        for listener in listeners:
             listener.notify(self.current_token)
-        self.listeners.clear()
 
     def remove(self, notifier):
         """ Remove this listener from all the indexes in the Notifier
@@ -202,7 +203,7 @@ class Notifier(object):
         user_streams = room_user_streams.copy()
 
         for user in extra_users:
-            user_stream = self.user_to_user_stream.get(user)
+            user_stream = self.user_to_user_stream.get(str(user))
             if user_stream is not None:
                 user_streams.add(user_stream)
 
@@ -288,12 +289,18 @@ class Notifier(object):
 
         timer = [None]
 
+        if result:
+            user_stream.listeners.discard(listener[0])
+            defer.returnValue(result)
+            return
+
         if timeout:
             timed_out = [False]
 
             def _timeout_listener():
                 timed_out[0] = True
                 timer[0] = None
+                user_stream.listeners.discard(listener[0])
                 listener[0].notify(from_token)
 
             # We create multiple notification listeners so we have to manage

From f1b83d88a3d3ad596631e51852a9802d0a7270a0 Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Wed, 13 May 2015 16:54:02 +0100
Subject: [PATCH 05/11] Discard unused NotifierUserStreams

---
 synapse/notifier.py                   | 50 ++++++++++++++++++---------
 tests/rest/client/v1/test_presence.py |  1 +
 tests/utils.py                        |  3 ++
 3 files changed, 38 insertions(+), 16 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index 6fcb7767a..344dd0317 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -71,14 +71,17 @@ class _NotifierUserStream(object):
     so that it can remove itself from the indexes in the Notifier class.
     """
 
-    def __init__(self, user, rooms, current_token, appservice=None):
+    def __init__(self, user, rooms, current_token, time_now_ms,
+                 appservice=None):
         self.user = str(user)
         self.appservice = appservice
         self.listeners = set()
         self.rooms = set(rooms)
         self.current_token = current_token
+        self.last_notified_ms = time_now_ms
 
-    def notify(self, stream_key, stream_id):
+    def notify(self, stream_key, stream_id, time_now_ms):
+        self.last_notified_ms = time_now_ms
         self.current_token = self.current_token.copy_and_replace(
             stream_key, stream_id
         )
@@ -96,7 +99,7 @@ class _NotifierUserStream(object):
             lst = notifier.room_to_user_streams.get(room, set())
             lst.discard(self)
 
-        notifier.user_to_user_streams.get(self.user, set()).discard(self)
+        notifier.user_to_user_stream.pop(self.user)
 
         if self.appservice:
             notifier.appservice_to_user_streams.get(
@@ -111,6 +114,8 @@ class Notifier(object):
     Primarily used from the /events stream.
     """
 
+    UNUSED_STREAM_EXPIRY_MS = 10 * 60 * 1000
+
     def __init__(self, hs):
         self.hs = hs
 
@@ -128,6 +133,10 @@ class Notifier(object):
             "user_joined_room", self._user_joined_room
         )
 
+        self.clock.looping_call(
+            self.remove_expired_streams, self.UNUSED_STREAM_EXPIRY_MS
+        )
+
         # This is not a very cheap test to perform, but it's only executed
         # when rendering the metrics page, which is likely once per minute at
         # most when scraping it.
@@ -221,9 +230,12 @@ class Notifier(object):
 
         logger.debug("on_new_room_event listeners %s", user_streams)
 
+        time_now_ms = self.clock.time_msec()
         for user_stream in user_streams:
             try:
-                user_stream.notify("room_key", "s%d" % (room_stream_id,))
+                user_stream.notify(
+                    "room_key", "s%d" % (room_stream_id,), time_now_ms
+                )
             except:
                 logger.exception("Failed to notify listener")
 
@@ -246,9 +258,10 @@ class Notifier(object):
         for room in rooms:
             user_streams |= self.room_to_user_streams.get(room, set())
 
+        time_now_ms = self.clock.time_msec()
         for user_stream in user_streams:
             try:
-                user_stream.notify(stream_key, new_token)
+                user_stream.notify(stream_key, new_token, time_now_ms)
             except:
                 logger.exception("Failed to notify listener")
 
@@ -260,6 +273,7 @@ class Notifier(object):
         """
 
         deferred = defer.Deferred()
+        time_now_ms = self.clock.time_msec()
 
         user = str(user)
         user_stream = self.user_to_user_stream.get(user)
@@ -272,6 +286,7 @@ class Notifier(object):
                 rooms=rooms,
                 appservice=appservice,
                 current_token=current_token,
+                time_now_ms=time_now_ms,
             )
             self._register_with_keys(user_stream)
         else:
@@ -365,6 +380,20 @@ class Notifier(object):
 
         defer.returnValue(result)
 
+    @log_function
+    def remove_expired_streams(self):
+        time_now_ms = self.clock.time_msec()
+        expired_streams = []
+        expire_before_ts = time_now_ms - self.UNUSED_STREAM_EXPIRY_MS
+        for stream in self.user_to_user_stream.values():
+            if stream.listeners:
+                continue
+            if stream.last_notified_ms < expire_before_ts:
+                expired_streams.append(stream)
+
+        for expired_stream in expired_streams:
+            expired_stream.remove(self)
+
     @log_function
     def _register_with_keys(self, user_stream):
         self.user_to_user_stream[user_stream.user] = user_stream
@@ -385,14 +414,3 @@ class Notifier(object):
             room_streams = self.room_to_user_streams.setdefault(room_id, set())
             room_streams.add(new_user_stream)
             new_user_stream.rooms.add(room_id)
-
-
-def _discard_if_notified(listener_set):
-    """Remove any 'stale' listeners from the given set.
-    """
-    to_discard = set()
-    for l in listener_set:
-        if l.notified():
-            to_discard.add(l)
-
-    listener_set -= to_discard
diff --git a/tests/rest/client/v1/test_presence.py b/tests/rest/client/v1/test_presence.py
index c0c52796a..29c0038f0 100644
--- a/tests/rest/client/v1/test_presence.py
+++ b/tests/rest/client/v1/test_presence.py
@@ -271,6 +271,7 @@ class PresenceEventStreamTestCase(unittest.TestCase):
                 "call_later",
                 "cancel_call_later",
                 "time_msec",
+                "looping_call",
             ]),
         )
 
diff --git a/tests/utils.py b/tests/utils.py
index a67530bd6..3b5c33591 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -197,6 +197,9 @@ class MockClock(object):
 
         return t
 
+    def looping_call(self, function, interval):
+        pass
+
     def cancel_call_later(self, timer):
         if timer[2]:
             raise Exception("Cannot cancel an expired timer")

From 5e0c533672f63990f7fdcf0b2c3cce15c3682d78 Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Wed, 13 May 2015 17:20:28 +0100
Subject: [PATCH 06/11] Fix metric counter

---
 synapse/notifier.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index 344dd0317..1f7f0a143 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -145,8 +145,8 @@ class Notifier(object):
 
             for x in self.room_to_user_streams.values():
                 all_user_streams |= x
-            for x in self.user_to_user_streams.values():
-                all_user_streams |= x
+            for x in self.user_to_user_stream:
+                all_user_streams.add(x)
             for x in self.appservice_to_user_streams.values():
                 all_user_streams |= x
 

From 3edd2d5c93ccbec46f101e65c6c7874a90bf0018 Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Thu, 14 May 2015 11:25:30 +0100
Subject: [PATCH 07/11] Fix v2 sync, update the last_notified_ms only if there
 was an active listener

---
 synapse/handlers/sync.py |  2 +-
 synapse/notifier.py      | 11 ++++++-----
 2 files changed, 7 insertions(+), 6 deletions(-)

diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 35a62fda4..bd8c60368 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -92,7 +92,7 @@ class SyncHandler(BaseHandler):
             result = yield self.current_sync_for_user(sync_config, since_token)
             defer.returnValue(result)
         else:
-            def current_sync_callback():
+            def current_sync_callback(before_token, after_token):
                 return self.current_sync_for_user(sync_config, since_token)
 
             rm_handler = self.hs.get_handlers().room_member_handler
diff --git a/synapse/notifier.py b/synapse/notifier.py
index 1f7f0a143..2de7dca8a 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -81,14 +81,15 @@ class _NotifierUserStream(object):
         self.last_notified_ms = time_now_ms
 
     def notify(self, stream_key, stream_id, time_now_ms):
-        self.last_notified_ms = time_now_ms
         self.current_token = self.current_token.copy_and_replace(
             stream_key, stream_id
         )
-        listeners = self.listeners
-        self.listeners = set()
-        for listener in listeners:
-            listener.notify(self.current_token)
+        if self.listeners:
+            self.last_notified_ms = time_now_ms
+            listeners = self.listeners
+            self.listeners = set()
+            for listener in listeners:
+                listener.notify(self.current_token)
 
     def remove(self, notifier):
         """ Remove this listener from all the indexes in the Notifier

From 084c365c3ad7f3a30ecfccca6cc79fb2e0c652ad Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Thu, 14 May 2015 12:03:26 +0100
Subject: [PATCH 08/11] Use the current token when timing out a notifier, make
 sure the user_id is a string in on_new_user_event

---
 synapse/notifier.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index 2de7dca8a..3dbd6f984 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -252,7 +252,7 @@ class Notifier(object):
         user_streams = set()
 
         for user in users:
-            user_stream = self.user_to_user_stream.get(user)
+            user_stream = self.user_to_user_stream.get(str(user))
             if user_stream is not None:
                 user_streams.add(user_stream)
 
@@ -317,7 +317,7 @@ class Notifier(object):
                 timed_out[0] = True
                 timer[0] = None
                 user_stream.listeners.discard(listener[0])
-                listener[0].notify(from_token)
+                listener[0].notify(current_token)
 
             # We create multiple notification listeners so we have to manage
             # canceling the timeout ourselves.

From 0ad1c672344e09a2a09690a1ab1bb61fe0397293 Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Thu, 14 May 2015 14:35:07 +0100
Subject: [PATCH 09/11] Add some doc-strings to notifier

---
 synapse/notifier.py | 50 +++++++++++++++++++++++++++++----------------
 1 file changed, 32 insertions(+), 18 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index 3dbd6f984..862b42cfc 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -81,6 +81,13 @@ class _NotifierUserStream(object):
         self.last_notified_ms = time_now_ms
 
     def notify(self, stream_key, stream_id, time_now_ms):
+        """Notify any listeners for this user of a new event from an
+        event source.
+        Args:
+            stream_key(str): The stream the event came from.
+            stream_id(str): The new id for the stream the event came from.
+            time_now_ms(int): The current time in milliseconds.
+        """
         self.current_token = self.current_token.copy_and_replace(
             stream_key, stream_id
         )
@@ -167,17 +174,6 @@ class Notifier(object):
             lambda: count(bool, self.appservice_to_user_streams.values()),
         )
 
-    def notify_pending_new_room_events(self, max_room_stream_id):
-        pending = sorted(self.pending_new_room_events)
-        self.pending_new_room_events = []
-        for event, room_stream_id, extra_users in pending:
-            if room_stream_id > max_room_stream_id:
-                self.pending_new_room_events.append((
-                    event, room_stream_id, extra_users
-                ))
-            else:
-                self._on_new_room_event(event, room_stream_id, extra_users)
-
     @log_function
     @defer.inlineCallbacks
     def on_new_room_event(self, event, room_stream_id, max_room_stream_id,
@@ -188,19 +184,37 @@ class Notifier(object):
         This triggers the notifier to wake up any listeners that are
         listening to the room, and any listeners for the users in the
         `extra_users` param.
+
+        The events can be peristed out of order. The notifier will wait
+        until all previous events have been persisted before notifying
+        the client streams.
         """
         yield run_on_reactor()
 
-        self.notify_pending_new_room_events(max_room_stream_id)
+        self.pending_new_room_events.append((
+            event, room_stream_id, extra_users
+        ))
+        self._notify_pending_new_room_events(max_room_stream_id)
 
-        if room_stream_id > max_room_stream_id:
-            self.pending_new_room_events.append((
-                event, room_stream_id, extra_users
-            ))
-        else:
-            self._on_new_room_event(event, room_stream_id, extra_users)
+    def _notify_pending_new_room_events(self, max_room_stream_id):
+        """Notify for the room events that were queued waiting for a previous
+        event to be persisted.
+        Args:
+            max_room_stream_id(int): The highest stream_id below which all
+                events have been persisted.
+        """
+        pending = sorted(self.pending_new_room_events)
+        self.pending_new_room_events = []
+        for event, room_stream_id, extra_users in pending:
+            if room_stream_id > max_room_stream_id:
+                self.pending_new_room_events.append((
+                    event, room_stream_id, extra_users
+                ))
+            else:
+                self._on_new_room_event(event, room_stream_id, extra_users)
 
     def _on_new_room_event(self, event, room_stream_id, extra_users=[]):
+        """Notify any user streams that are interested in this room event"""
         # poke any interested application service.
         self.hs.get_handlers().appservice_handler.notify_interested_services(
             event

From 1e90715a3d5f8a910c187dec888283e110a3c04a Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Mon, 18 May 2015 13:17:36 +0100
Subject: [PATCH 10/11] Make sure the notifier stream token goes forward when
 it is updated. Sort the pending events by the correct room_stream_id

---
 synapse/notifier.py |  8 ++++----
 synapse/types.py    | 17 +++++++++++++++++
 2 files changed, 21 insertions(+), 4 deletions(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index 862b42cfc..0b5d97521 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -88,7 +88,7 @@ class _NotifierUserStream(object):
             stream_id(str): The new id for the stream the event came from.
             time_now_ms(int): The current time in milliseconds.
         """
-        self.current_token = self.current_token.copy_and_replace(
+        self.current_token = self.current_token.copy_and_advance(
             stream_key, stream_id
         )
         if self.listeners:
@@ -192,7 +192,7 @@ class Notifier(object):
         yield run_on_reactor()
 
         self.pending_new_room_events.append((
-            event, room_stream_id, extra_users
+            room_stream_id, event, extra_users
         ))
         self._notify_pending_new_room_events(max_room_stream_id)
 
@@ -205,10 +205,10 @@ class Notifier(object):
         """
         pending = sorted(self.pending_new_room_events)
         self.pending_new_room_events = []
-        for event, room_stream_id, extra_users in pending:
+        for room_stream_id, event, extra_users in pending:
             if room_stream_id > max_room_stream_id:
                 self.pending_new_room_events.append((
-                    event, room_stream_id, extra_users
+                    room_stream_id, event, extra_users
                 ))
             else:
                 self._on_new_room_event(event, room_stream_id, extra_users)
diff --git a/synapse/types.py b/synapse/types.py
index d89a04f7c..1b21160c5 100644
--- a/synapse/types.py
+++ b/synapse/types.py
@@ -119,6 +119,7 @@ class StreamToken(
     @property
     def room_stream_id(self):
         # TODO(markjh): Awful hack to work around hacks in the presence tests
+        # which assume that the keys are integers.
         if type(self.room_key) is int:
             return self.room_key
         else:
@@ -132,6 +133,22 @@ class StreamToken(
             or (int(other_token.typing_key) < int(self.typing_key))
         )
 
+    def copy_and_advance(self, key, new_value):
+        """Advance the given key in the token to a new value if and only if the
+        new value is after the old value.
+        """
+        new_token = self.copy_and_replace(key, new_value)
+        if key == "room_key":
+            new_id = new_token.room_stream_id
+            old_id = self.room_stream_id
+        else:
+            new_id = int(getattr(new_token, key))
+            old_id = int(getattr(self, key))
+        if old_id < new_id:
+            return new_token
+        else:
+            return self
+
     def copy_and_replace(self, key, new_value):
         d = self._asdict()
         d[key] = new_value

From ad31fa304022e593b6c9354d3a42df761597e69b Mon Sep 17 00:00:00 2001
From: Mark Haines <mark.haines@matrix.org>
Date: Mon, 18 May 2015 14:04:58 +0100
Subject: [PATCH 11/11] Don't bother sorting by the room_stream_ids, it
 shouldn't matter which order they are notified in

---
 synapse/notifier.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/synapse/notifier.py b/synapse/notifier.py
index 0b5d97521..1e73d52c4 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -203,7 +203,7 @@ class Notifier(object):
             max_room_stream_id(int): The highest stream_id below which all
                 events have been persisted.
         """
-        pending = sorted(self.pending_new_room_events)
+        pending = self.pending_new_room_events
         self.pending_new_room_events = []
         for room_stream_id, event, extra_users in pending:
             if room_stream_id > max_room_stream_id: