mirrors/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2024-11-23 10:05:55 +03:00
Code Issues Projects Releases Packages Wiki Activity

Handle pusher being deleted during processing.

Browse source 
Instead of throwing a StoreError lets break out of processing loop and
mark the pusher as stopped.
This commit is contained in:
Erik Johnston 2019-08-01 13:41:27 +01:00
parent 58af30a6c7
commit d02e41dcb2
3 changed files with 55 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
synapse/push/emailpusher.py
View file

@ -234,6 +234,7 @@ class EmailPusher(object):
return
self.last_stream_ordering = last_stream_ordering
pusher_still_exists = (
yield self.store.update_pusher_last_stream_ordering_and_success(
self.app_id,
self.email,
@ -241,6 +242,12 @@ class EmailPusher(object):
last_stream_ordering,
self.clock.time_msec(),
)
)
if not pusher_still_exists:
# The pusher has been deleted while we were processing, so
# lets just stop and return.
self.on_stop()
return
def seconds_until(self, ts_msec):
secs = (ts_msec - self.clock.time_msec()) / 1000

15
synapse/push/httppusher.py
View file

@ -199,6 +199,7 @@ class HttpPusher(object):
http_push_processed_counter.inc()
self.backoff_delay = HttpPusher.INITIAL_BACKOFF_SEC
self.last_stream_ordering = push_action["stream_ordering"]
pusher_still_exists = (
yield self.store.update_pusher_last_stream_ordering_and_success(
self.app_id,
self.pushkey,
@ -206,6 +207,13 @@ class HttpPusher(object):
self.last_stream_ordering,
self.clock.time_msec(),
)
)
if not pusher_still_exists:
# The pusher has been deleted while we were processing, so
# lets just stop and return.
self.on_stop()
return
if self.failing_since:
self.failing_since = None
yield self.store.update_pusher_failing_since(
@ -234,12 +242,17 @@ class HttpPusher(object):
)
self.backoff_delay = HttpPusher.INITIAL_BACKOFF_SEC
self.last_stream_ordering = push_action["stream_ordering"]
yield self.store.update_pusher_last_stream_ordering(
pusher_still_exists = yield self.store.update_pusher_last_stream_ordering(
self.app_id,
self.pushkey,
self.user_id,
self.last_stream_ordering,
)
if not pusher_still_exists:
# The pusher has been deleted while we were processing, so
# lets just stop and return.
self.on_stop()
return
self.failing_since = None
yield self.store.update_pusher_failing_since(

30
synapse/storage/pusher.py
View file

@ -308,22 +308,36 @@ class PusherStore(PusherWorkerStore):
def update_pusher_last_stream_ordering_and_success(
self, app_id, pushkey, user_id, last_stream_ordering, last_success
):
yield self._simple_update_one(
"pushers",
{"app_id": app_id, "pushkey": pushkey, "user_name": user_id},
{
"""Update the last stream ordering position we've processed up to for
the given pusher.
Args:
app_id (str)
pushkey (str)
last_stream_ordering (int)
last_success (int)
Returns:
Deferred[bool]: Whether the pusher stil exists or not.
"""
updated = yield self._simple_update(
table="pushers",
keyvalues={"app_id": app_id, "pushkey": pushkey, "user_name": user_id},
updatevalues={
"last_stream_ordering": last_stream_ordering,
"last_success": last_success,
},
desc="update_pusher_last_stream_ordering_and_success",
)
return bool(updated)
@defer.inlineCallbacks
def update_pusher_failing_since(self, app_id, pushkey, user_id, failing_since):
yield self._simple_update_one(
"pushers",
{"app_id": app_id, "pushkey": pushkey, "user_name": user_id},
{"failing_since": failing_since},
yield self._simple_update(
table="pushers",
keyvalues={"app_id": app_id, "pushkey": pushkey, "user_name": user_id},
updatevalues={"failing_since": failing_since},
desc="update_pusher_failing_since",
)