diff options
Diffstat (limited to 'synapse/handlers')
| -rw-r--r-- | synapse/handlers/events.py | 11 | ||||
| -rw-r--r-- | synapse/handlers/sync.py | 16 |
2 files changed, 20 insertions, 7 deletions
diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index fe300433e6..576d77e0e7 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -69,7 +69,12 @@ class EventStreamHandler(BaseHandler): A deferred that completes once their presence has been updated. """ if user not in self._streams_per_user: - self._streams_per_user[user] = 0 + # Make sure we set the streams per user to 1 here rather than + # setting it to zero and incrementing the value below. + # Otherwise this may race with stopped_stream causing the + # user to be erased from the map before we have a chance + # to increment it. + self._streams_per_user[user] = 1 if user in self._stop_timer_per_user: try: self.clock.cancel_call_later( @@ -79,8 +84,8 @@ class EventStreamHandler(BaseHandler): logger.exception("Failed to cancel event timer") else: yield started_user_eventstream(self.distributor, user) - - self._streams_per_user[user] += 1 + else: + self._streams_per_user[user] += 1 def stopped_stream(self, user): """If there are no streams for a user this starts a timer that will diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 24c2b2fad6..7088c20cb4 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -17,6 +17,7 @@ from ._base import BaseHandler from synapse.streams.config import PaginationConfig from synapse.api.constants import Membership, EventTypes +from synapse.util import unwrapFirstError from twisted.internet import defer @@ -209,9 +210,10 @@ class SyncHandler(BaseHandler): joined = [] invited = [] archived = [] + deferreds = [] for event in room_list: if event.membership == Membership.JOIN: - room_sync = yield self.full_state_sync_for_joined_room( + room_sync_deferred = self.full_state_sync_for_joined_room( room_id=event.room_id, sync_config=sync_config, now_token=now_token, @@ -220,7 +222,8 @@ class SyncHandler(BaseHandler): tags_by_room=tags_by_room, account_data_by_room=account_data_by_room, ) - joined.append(room_sync) + room_sync_deferred.addCallback(joined.append) + deferreds.append(room_sync_deferred) elif event.membership == Membership.INVITE: invite = yield self.store.get_event(event.event_id) invited.append(InvitedSyncResult( @@ -231,7 +234,7 @@ class SyncHandler(BaseHandler): leave_token = now_token.copy_and_replace( "room_key", "s%d" % (event.stream_ordering,) ) - room_sync = yield self.full_state_sync_for_archived_room( + room_sync_deferred = self.full_state_sync_for_archived_room( sync_config=sync_config, room_id=event.room_id, leave_event_id=event.event_id, @@ -240,7 +243,12 @@ class SyncHandler(BaseHandler): tags_by_room=tags_by_room, account_data_by_room=account_data_by_room, ) - archived.append(room_sync) + room_sync_deferred.addCallback(archived.append) + deferreds.append(room_sync_deferred) + + yield defer.gatherResults( + deferreds, consumeErrors=True + ).addErrback(unwrapFirstError) defer.returnValue(SyncResult( presence=presence, |