diff --git a/synapse/replication/slave/storage/deviceinbox.py b/synapse/replication/slave/storage/deviceinbox.py
index 64d8eb2af1..251078ba57 100644
--- a/synapse/replication/slave/storage/deviceinbox.py
+++ b/synapse/replication/slave/storage/deviceinbox.py
@@ -16,6 +16,7 @@
from ._base import BaseSlavedStore
from ._slaved_id_tracker import SlavedIdTracker
from synapse.storage import DataStore
+from synapse.util.caches.stream_change_cache import StreamChangeCache
class SlavedDeviceInboxStore(BaseSlavedStore):
@@ -24,6 +25,10 @@ class SlavedDeviceInboxStore(BaseSlavedStore):
self._device_inbox_id_gen = SlavedIdTracker(
db_conn, "device_inbox", "stream_id",
)
+ self._device_inbox_stream_cache = StreamChangeCache(
+ "DeviceInboxStreamChangeCache",
+ self._device_inbox_id_gen.get_current_token()
+ )
get_to_device_stream_token = DataStore.get_to_device_stream_token.__func__
get_new_messages_for_device = DataStore.get_new_messages_for_device.__func__
@@ -38,5 +43,11 @@ class SlavedDeviceInboxStore(BaseSlavedStore):
stream = result.get("to_device")
if stream:
self._device_inbox_id_gen.advance(int(stream["position"]))
+ for row in stream["rows"]:
+ stream_id = row[0]
+ user_id = row[1]
+ self._device_inbox_stream_cache.entity_has_changed(
+ user_id, stream_id
+ )
return super(SlavedDeviceInboxStore, self).process_replication(result)
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 6965daddc5..828e5ca60b 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -202,7 +202,7 @@ class DataStore(RoomMemberStore, RoomStore,
max_value=max_device_inbox_id,
)
self._device_federation_outbox_stream_cache = StreamChangeCache(
- "DeviceInboxStreamChangeCache", min_device_outbox_id,
+ "DeviceFederationOutboxStreamChangeCache", min_device_outbox_id,
prefilled_cache=device_outbox_prefill,
)
|
