summary refs log tree commit diff
path: root/synapse/replication
diff options
context:
space:
mode:
authorMark Haines <mark.haines@matrix.org>2016-05-13 17:33:44 +0100
committerMark Haines <mark.haines@matrix.org>2016-05-13 17:33:44 +0100
commit0466454b003860dba23363f882916eb4f7d27648 (patch)
tree2025383a5c338dd73b06e7e911ef8ef5a365af05 /synapse/replication
parentMerge pull request #780 from matrix-org/dbkr/email_notifs_on_pusher (diff)
downloadsynapse-0466454b003860dba23363f882916eb4f7d27648.tar.xz
Assert that stream replicated stream positions are ints
Diffstat (limited to 'synapse/replication')
-rw-r--r--synapse/replication/slave/storage/events.py8
-rw-r--r--synapse/replication/slave/storage/pushers.py4
-rw-r--r--synapse/replication/slave/storage/receipts.py2
3 files changed, 7 insertions, 7 deletions
diff --git a/synapse/replication/slave/storage/events.py b/synapse/replication/slave/storage/events.py
index 99cddf2518..c0d741452d 100644
--- a/synapse/replication/slave/storage/events.py
+++ b/synapse/replication/slave/storage/events.py
@@ -149,7 +149,7 @@ class SlavedEventStore(BaseSlavedStore):
 
         stream = result.get("events")
         if stream:
-            self._stream_id_gen.advance(stream["position"])
+            self._stream_id_gen.advance(int(stream["position"]))
             for row in stream["rows"]:
                 self._process_replication_row(
                     row, backfilled=False, state_resets=state_resets
@@ -157,7 +157,7 @@ class SlavedEventStore(BaseSlavedStore):
 
         stream = result.get("backfill")
         if stream:
-            self._backfill_id_gen.advance(-stream["position"])
+            self._backfill_id_gen.advance(-int(stream["position"]))
             for row in stream["rows"]:
                 self._process_replication_row(
                     row, backfilled=True, state_resets=state_resets
@@ -165,14 +165,14 @@ class SlavedEventStore(BaseSlavedStore):
 
         stream = result.get("forward_ex_outliers")
         if stream:
-            self._stream_id_gen.advance(stream["position"])
+            self._stream_id_gen.advance(int(stream["position"]))
             for row in stream["rows"]:
                 event_id = row[1]
                 self._invalidate_get_event_cache(event_id)
 
         stream = result.get("backward_ex_outliers")
         if stream:
-            self._backfill_id_gen.advance(-stream["position"])
+            self._backfill_id_gen.advance(-int(stream["position"]))
             for row in stream["rows"]:
                 event_id = row[1]
                 self._invalidate_get_event_cache(event_id)
diff --git a/synapse/replication/slave/storage/pushers.py b/synapse/replication/slave/storage/pushers.py
index 8faddb2595..d88206b3bb 100644
--- a/synapse/replication/slave/storage/pushers.py
+++ b/synapse/replication/slave/storage/pushers.py
@@ -43,10 +43,10 @@ class SlavedPusherStore(BaseSlavedStore):
     def process_replication(self, result):
         stream = result.get("pushers")
         if stream:
-            self._pushers_id_gen.advance(stream["position"])
+            self._pushers_id_gen.advance(int(stream["position"]))
 
         stream = result.get("deleted_pushers")
         if stream:
-            self._pushers_id_gen.advance(stream["position"])
+            self._pushers_id_gen.advance(int(stream["position"]))
 
         return super(SlavedPusherStore, self).process_replication(result)
diff --git a/synapse/replication/slave/storage/receipts.py b/synapse/replication/slave/storage/receipts.py
index b55d5dfd08..ec007516d0 100644
--- a/synapse/replication/slave/storage/receipts.py
+++ b/synapse/replication/slave/storage/receipts.py
@@ -50,7 +50,7 @@ class SlavedReceiptsStore(BaseSlavedStore):
     def process_replication(self, result):
         stream = result.get("receipts")
         if stream:
-            self._receipts_id_gen.advance(stream["position"])
+            self._receipts_id_gen.advance(int(stream["position"]))
             for row in stream["rows"]:
                 room_id, receipt_type, user_id = row[1:4]
                 self.invalidate_caches_for_receipt(room_id, receipt_type, user_id)