summary refs log tree commit diff
path: root/synapse/storage/_base.py
diff options
context:
space:
mode:
authorErik Johnston <erikj@jki.re>2019-02-27 10:28:37 +0000
committerRichard van der Hoff <1389908+richvdh@users.noreply.github.com>2019-02-27 10:28:37 +0000
commit6bb1c028f190d8ba561a6deaa474c060efb7c502 (patch)
tree3a944f6476fc017cdc93143bbdac3c6af81dffc7 /synapse/storage/_base.py
parentAdd database version to phonehome stats. (#4753) (diff)
downloadsynapse-6bb1c028f190d8ba561a6deaa474c060efb7c502.tar.xz
Limit cache invalidation replication line length (#4748)
Diffstat (limited to 'synapse/storage/_base.py')
-rw-r--r--synapse/storage/_base.py15
1 files changed, 11 insertions, 4 deletions
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 190be34fb1..a0333d5309 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -30,6 +30,7 @@ from synapse.api.errors import StoreError
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.engines import PostgresEngine, Sqlite3Engine
 from synapse.types import get_domain_from_id
+from synapse.util import batch_iter
 from synapse.util.caches.descriptors import Cache
 from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
 from synapse.util.stringutils import exception_to_unicode
@@ -1327,10 +1328,16 @@ class SQLBaseStore(object):
         """
         txn.call_after(self._invalidate_state_caches, room_id, members_changed)
 
-        keys = itertools.chain([room_id], members_changed)
-        self._send_invalidation_to_replication(
-            txn, _CURRENT_STATE_CACHE_NAME, keys,
-        )
+        # We need to be careful that the size of the `members_changed` list
+        # isn't so large that it causes problems sending over replication, so we
+        # send them in chunks.
+        # Max line length is 16K, and max user ID length is 255, so 50 should
+        # be safe.
+        for chunk in batch_iter(members_changed, 50):
+            keys = itertools.chain([room_id], chunk)
+            self._send_invalidation_to_replication(
+                txn, _CURRENT_STATE_CACHE_NAME, keys,
+            )
 
     def _invalidate_state_caches(self, room_id, members_changed):
         """Invalidates caches that are based on the current state, but does