summary refs log tree commit diff
path: root/synapse/replication/slave/storage/devices.py
diff options
context:
space:
mode:
authorHubert Chathi <hubert@uhoreg.ca>2019-10-30 17:22:52 -0400
committerHubert Chathi <hubert@uhoreg.ca>2019-10-30 17:22:52 -0400
commit998f7fe7d4ddb2dddf4d46a8a420a6fd7e37577c (patch)
tree3129850a4e62fd6583aa5f85d2b209c1bdccf80e /synapse/replication/slave/storage/devices.py
parentMerge branch 'develop' into uhoreg/cross_signing_fix_workers_notify (diff)
downloadsynapse-998f7fe7d4ddb2dddf4d46a8a420a6fd7e37577c.tar.xz
make user signatures a separate stream
Diffstat (limited to 'synapse/replication/slave/storage/devices.py')
-rw-r--r--synapse/replication/slave/storage/devices.py8
1 files changed, 6 insertions, 2 deletions
diff --git a/synapse/replication/slave/storage/devices.py b/synapse/replication/slave/storage/devices.py
index c9f99a8405..f416d73b2e 100644
--- a/synapse/replication/slave/storage/devices.py
+++ b/synapse/replication/slave/storage/devices.py
@@ -42,7 +42,9 @@ class SlavedDeviceStore(EndToEndKeyWorkerStore, DeviceWorkerStore, BaseSlavedSto
 
     def stream_positions(self):
         result = super(SlavedDeviceStore, self).stream_positions()
-        result["device_lists"] = self._device_list_id_gen.get_current_token()
+        result["user_signature"] = result[
+            "device_lists"
+        ] = self._device_list_id_gen.get_current_token()
         return result
 
     def process_replication_rows(self, stream_name, token, rows):
@@ -50,13 +52,15 @@ class SlavedDeviceStore(EndToEndKeyWorkerStore, DeviceWorkerStore, BaseSlavedSto
             self._device_list_id_gen.advance(token)
             for row in rows:
                 self._invalidate_caches_for_devices(token, row.user_id, row.destination)
+        elif stream_name == "user_signature":
+            for row in rows:
+                self._user_signature_stream_cache.entity_has_changed(row.user_id, token)
         return super(SlavedDeviceStore, self).process_replication_rows(
             stream_name, token, rows
         )
 
     def _invalidate_caches_for_devices(self, token, user_id, destination):
         self._device_list_stream_cache.entity_has_changed(user_id, token)
-        self._user_signature_stream_cache.entity_has_changed(user_id, token)
 
         if destination:
             self._device_list_federation_stream_cache.entity_has_changed(