MirrorHub/synapse
0
0
Fork 1
mirror of https://mau.dev/maunium/synapse.git synced 2024-11-18 16:02:15 +01:00
Code Issues Projects Releases Wiki Activity

Run unsafe proces in a loop until we've caught up

Browse source 
and wrap unsafe process in a try block
This commit is contained in:
David Baker 2016-04-11 12:48:30 +01:00
parent d96a070a3a
commit 9bb041791c
1 changed files with 11 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
synapse/push/httppusher.py
View file

@ -116,7 +116,16 @@ class HttpPusher(object):
return return
try: try:
self.processing = True self.processing = True
yield self._unsafe_process() # if the max ordering changes while we're running _unsafe_process,
# call it again, and so on until we've caught up.
while True:
starting_max_ordering = self.max_stream_ordering
try:
yield self._unsafe_process()
except:
logger.exception("Exception processing notifs")
if self.max_stream_ordering == starting_max_ordering:
break
finally: finally:
self.processing = False self.processing = False
@ -127,7 +136,7 @@ class HttpPusher(object):
Never call this directly: use _process which will only allow this to Never call this directly: use _process which will only allow this to
run once per pusher. run once per pusher.
""" """
starting_max_ordering = self.max_stream_ordering
unprocessed = yield self.store.get_unread_push_actions_for_user_in_range( unprocessed = yield self.store.get_unread_push_actions_for_user_in_range(
self.user_id, self.last_stream_ordering, self.max_stream_ordering self.user_id, self.last_stream_ordering, self.max_stream_ordering
) )
@ -188,8 +197,6 @@ class HttpPusher(object):
self.timed_call = reactor.callLater(self.backoff_delay, self.on_timer) self.timed_call = reactor.callLater(self.backoff_delay, self.on_timer)
self.backoff_delay = min(self.backoff_delay * 2, self.MAX_BACKOFF_SEC) self.backoff_delay = min(self.backoff_delay * 2, self.MAX_BACKOFF_SEC)
break break
if self.max_stream_ordering != starting_max_ordering:
self._unsafe_process()
@defer.inlineCallbacks @defer.inlineCallbacks
def _process_one(self, push_action): def _process_one(self, push_action):