diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py
index f2013faeb2..8fcb8ac3d9 100644
--- a/synapse/rest/client/sync.py
+++ b/synapse/rest/client/sync.py
@@ -16,7 +16,7 @@ import logging
from collections import defaultdict
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
-from synapse.api.constants import EduTypes, Membership, PresenceState
+from synapse.api.constants import AccountDataTypes, EduTypes, Membership, PresenceState
from synapse.api.errors import Codes, StoreError, SynapseError
from synapse.api.filtering import FilterCollection
from synapse.api.presence import UserPresenceState
@@ -139,7 +139,28 @@ class SyncRestServlet(RestServlet):
device_id,
)
- request_key = (user, timeout, since, filter_id, full_state, device_id)
+ # Stream position of the last ignored users account data event for this user,
+ # if we're initial syncing.
+ # We include this in the request key to invalidate an initial sync
+ # in the response cache once the set of ignored users has changed.
+ # (We filter out ignored users from timeline events, so our sync response
+ # is invalid once the set of ignored users changes.)
+ last_ignore_accdata_streampos: Optional[int] = None
+ if not since:
+ # No `since`, so this is an initial sync.
+ last_ignore_accdata_streampos = await self.store.get_latest_stream_id_for_global_account_data_by_type_for_user(
+ user.to_string(), AccountDataTypes.IGNORED_USER_LIST
+ )
+
+ request_key = (
+ user,
+ timeout,
+ since,
+ filter_id,
+ full_state,
+ device_id,
+ last_ignore_accdata_streampos,
+ )
if filter_id is None:
filter_collection = self.filtering.DEFAULT_FILTER_COLLECTION
diff --git a/synapse/storage/databases/main/account_data.py b/synapse/storage/databases/main/account_data.py
index 95567826f2..308d19440f 100644
--- a/synapse/storage/databases/main/account_data.py
+++ b/synapse/storage/databases/main/account_data.py
@@ -237,6 +237,37 @@ class AccountDataWorkerStore(PushRulesWorkerStore, CacheInvalidationWorkerStore)
else:
return None
+ async def get_latest_stream_id_for_global_account_data_by_type_for_user(
+ self, user_id: str, data_type: str
+ ) -> Optional[int]:
+ """
+ Returns:
+ The stream ID of the account data,
+ or None if there is no such account data.
+ """
+
+ def get_latest_stream_id_for_global_account_data_by_type_for_user_txn(
+ txn: LoggingTransaction,
+ ) -> Optional[int]:
+ sql = """
+ SELECT stream_id FROM account_data
+ WHERE user_id = ? AND account_data_type = ?
+ ORDER BY stream_id DESC
+ LIMIT 1
+ """
+ txn.execute(sql, (user_id, data_type))
+
+ row = txn.fetchone()
+ if row:
+ return row[0]
+ else:
+ return None
+
+ return await self.db_pool.runInteraction(
+ "get_latest_stream_id_for_global_account_data_by_type_for_user",
+ get_latest_stream_id_for_global_account_data_by_type_for_user_txn,
+ )
+
@cached(num_args=2, tree=True)
async def get_account_data_for_room(
self, user_id: str, room_id: str
|
