diff --git a/changelog.d/20289.bugfix b/changelog.d/20289.bugfix new file mode 100644 index 0000000000..74b7d1f0aa --- /dev/null +++ b/changelog.d/20289.bugfix @@ -0,0 +1 @@ +Fix bug where membership events would not be returned in `/sync` when using lazy-loaded members and using an old token for `since`. diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index a3ebf8bb1d..2eb5eb79c6 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -23,6 +23,7 @@ import itertools import json import logging import os +from collections import OrderedDict from typing import ( TYPE_CHECKING, AbstractSet, @@ -74,6 +75,7 @@ from synapse.types import ( MutableStateMap, Requester, RoomStreamToken, + StateKey, StateMap, StrCollection, StreamKeyType, @@ -130,6 +132,96 @@ LAZY_LOADED_PROFILE_FIELDS_CACHE_DIGEST_SIZE = 16 LAZY_LOADED_PROFILE_FIELDS_CACHE_DIGEST_KEY = os.urandom(32) +@attr.s(slots=True, auto_attribs=True) +class LazyLoadedMembersCache: + """The membership events a lazy-loading client has been sent, so that later + syncs can leave them out. One per (user, device). + + A member may only be left out if the client processed the response that + carried it. Each entry therefore records the `next_batch` position of the + response that carried it. A client that has processed a response syncs + from its `next_batch` next, so a request from `since` means the client + processed every response up to `since` and none after it. `start_request` + forgets the entries recorded after `since`. They were in responses the + client lost, by timing out a slow request and retrying it (#19978) or by + reloading from an older persisted token (#20278). Entries recorded at or + before `since` stay, so a client resuming from a token a few minutes old is + only sent again what it lost in those minutes. An initial sync forgets + everything. + + Forgetting everything whenever `since` is before the newest entry would be + simpler, and `_newest` alone would suffice. But Element Web only persists + its sync token every few minutes and resumes from it on every reload, so + that would re-send every cached member on each reload. + + The notifier may build several results for one long-poll and returns only + the first non-empty one. An empty result carries no members, so it records + nothing. + + Leaving out a member the client never received loses it for good. A + `state_after` client does not apply state from the timeline, and a `state` + client only sees the membership changes that fall inside the timeline. + """ + + # user_id -> (event_id of the membership we last sent, room position of the + # `next_batch` of the response that carried it). Least recently used first. + # + # Not an `LruCache`, as that cannot be iterated over and `start_request` + # needs to check every entry's position against `since`. + _sent: OrderedDict[str, tuple[str, RoomStreamToken]] = attr.Factory(OrderedDict) + + # An upper bound on the positions of the entries, so that `start_request` + # need not look at them when nothing was recorded after `since`. That is the + # steady state, where `since` is the `next_batch` of the last response. + _newest: RoomStreamToken | None = None + + def start_request(self, since_token: StreamToken | None) -> None: + """Forget the members the client did not receive. Called at the start of + each sync request for this device.""" + if since_token is None: + self._sent.clear() + self._newest = None + return + + # Check for the common case where nothing has been recorded after + # `since`. + since = since_token.room_key + if self._newest is None or self._newest.is_before_or_eq(since): + return + + # Remove all entries that were sent after `since`. This only has + # `LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE` entries. + for user_id, (_, sent_at) in list(self._sent.items()): + if not sent_at.is_before_or_eq(since): + del self._sent[user_id] + + # Everything left is at or before `since`. + self._newest = since + + def was_sent(self, user_id: str, event_id: str) -> bool: + """Whether `event_id` is the membership of `user_id` we last sent.""" + entry = self._sent.get(user_id) + if entry is None or entry[0] != event_id: + return False + self._sent.move_to_end(user_id) + return True + + def mark_sent(self, user_id: str, event_id: str, next_batch: StreamToken) -> None: + """Record that `event_id` is being sent in the response ending at + `next_batch`.""" + sent_at = next_batch.room_key + self._sent[user_id] = (event_id, sent_at) + self._sent.move_to_end(user_id) + while len(self._sent) > LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE: + self._sent.popitem(last=False) + + # Only move the bound forwards. Positions taken by this process only + # ever advance, but a response that started earlier may record its + # entries after a later one has. + if self._newest is None or self._newest.is_before_or_eq(sent_at): + self._newest = sent_at + + SyncRequestKey = tuple[Any, ...] @@ -345,9 +437,9 @@ class SyncHandler: timeout=hs.config.caches.sync_response_cache_duration, ) - # ExpiringCache((User, Device)) -> LruCache(user_id => event_id) + # ExpiringCache((User, Device)) -> LazyLoadedMembersCache self.lazy_loaded_members_cache: ExpiringCache[ - tuple[str, str | None], LruCache[str, str] + tuple[str, str | None], LazyLoadedMembersCache ] = ExpiringCache( cache_name="lazy_loaded_members_cache", server_name=self.server_name, @@ -1088,7 +1180,7 @@ class SyncHandler: member_ids[hero_id] for hero_id in summary["m.heroes"] if ( - cache.get(hero_id) != member_ids[hero_id] + not cache.was_sent(hero_id, member_ids[hero_id]) and hero_id not in existing_members ) ] @@ -1096,27 +1188,21 @@ class SyncHandler: missing_hero_state = await self.store.get_events(missing_hero_event_ids) for s in missing_hero_state.values(): - cache.set(s.state_key, s.event_id) + cache.mark_sent(s.state_key, s.event_id, now_token) state[(EventTypes.Member, s.state_key)] = s return summary def get_lazy_loaded_members_cache( self, cache_key: tuple[str, str | None] - ) -> LruCache[str, str]: - # FIXME: This cache may be subject to losing members in the case that - # a sync is interrupted and retried, see https://github.com/element-hq/synapse/issues/19978 - cache: LruCache[str, str] | None = self.lazy_loaded_members_cache.get(cache_key) + ) -> LazyLoadedMembersCache: + cache = self.lazy_loaded_members_cache.get(cache_key) if cache is None: - logger.debug("creating LruCache for %r", cache_key) - cache = LruCache( - max_size=LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE, - clock=self.clock, - server_name=self.server_name, - ) + logger.debug("creating LazyLoadedMembersCache for %r", cache_key) + cache = LazyLoadedMembersCache() self.lazy_loaded_members_cache[cache_key] = cache else: - logger.debug("found LruCache for %r", cache_key) + logger.debug("found LazyLoadedMembersCache for %r", cache_key) return cache def get_lazy_loaded_profile_fields_cache( @@ -1155,6 +1241,7 @@ class SyncHandler: sync_config: SyncConfig, since_token: StreamToken | None, end_token: StreamToken, + now_token: StreamToken, full_state: bool, joined: bool, ) -> MutableStateMap[EventBase]: @@ -1169,6 +1256,7 @@ class SyncHandler: end_token: Token of the end of the current batch. Normally this will be the same as the global "now_token", but if the user has left the room, the point just after their leave event. + now_token: The `next_batch` token of the response being built. full_state: Whether to force returning the full state. `lazy_load_members` still applies when `full_state` is `True`. joined: whether the user is currently joined to the room @@ -1254,6 +1342,11 @@ class SyncHandler: # sync's timeline and the start of the current sync's timeline. # See the docstring above for details. state_ids: StateMap[str] + # The keys in `state_ids` whose state changed since `since_token`. + # These are sent whether or not the lazy-loaded members cache says + # the client has them. The rest are memberships fetched for + # timeline senders, which the client may already have. + changed_keys: set[StateKey] = set() # We need to know whether the state we fetch may be partial, so check # whether the room is partial stated *before* fetching it. is_partial_state_room = await self.store.is_partial_state_room(room_id) @@ -1273,7 +1366,10 @@ class SyncHandler: # is indeed the case. assert since_token is not None - state_ids = await self._compute_state_delta_for_incremental_sync( + ( + state_ids, + changed_keys, + ) = await self._compute_state_delta_for_incremental_sync( room_id, sync_config, batch, @@ -1309,30 +1405,28 @@ class SyncHandler: cache_key = (sync_config.user.to_string(), sync_config.device_id) cache = self.get_lazy_loaded_members_cache(cache_key) - # if it's a new sync sequence, then assume the client has had - # amnesia and doesn't want any recent lazy-loaded members - # de-duplicated. - if since_token is None: - logger.debug("clearing LruCache for %r", cache_key) - cache.clear() - else: - # only send members which aren't in our LruCache (either - # because they're new to this client or have been pushed out - # of the cache) - logger.debug("filtering state from %r...", state_ids) - state_ids = { - t: event_id - for t, event_id in state_ids.items() - if cache.get(t[1]) != event_id - } - logger.debug("...to %r", state_ids) + # Only send members which aren't in our cache (either because + # they're new to this client or have been pushed out of the + # cache). For an initial sync the cache has already been + # cleared. + # + # A state change since `since_token` is always sent, regardless + # of whether it was previously sent to the client (e.g due to a + # client retrying the request). + logger.debug("filtering state from %r...", state_ids) + state_ids = { + t: event_id + for t, event_id in state_ids.items() + if t in changed_keys or not cache.was_sent(t[1], event_id) + } + logger.debug("...to %r", state_ids) - # add any member IDs we are about to send into our LruCache + # add any member IDs we are about to send into our cache for t, event_id in itertools.chain( state_ids.items(), timeline_state.items() ): if t[0] == EventTypes.Member: - cache.set(t[1], event_id) + cache.mark_sent(t[1], event_id, now_token) state: dict[str, EventBase] = {} if state_ids: @@ -1505,14 +1599,16 @@ class SyncHandler: members_to_fetch: set[str] | None, timeline_state: StateMap[str], joined: bool, - ) -> StateMap[str]: + ) -> tuple[StateMap[str], set[StateKey]]: """Calculate the state events to be included in an incremental sync response. If lazy-loading of membership events is enabled (as indicated by `members_to_fetch` being not-`None`), the result will include the membership events for each member in `members_to_fetch`. The caller (`compute_state_delta`) is responsible for keeping track of which membership - events we have already sent to the client, and hence ripping them out. + events we have already sent to the client, and hence ripping them out. It + may only rip out entries that are not state changes, so this also returns + the keys whose state changed since `since_token`. Note that whether this returns the state at the start or the end of the batch depends on `sync_config.use_state_after` (c.f. MSC4222). @@ -1533,7 +1629,10 @@ class SyncHandler: Returns: A map from (type, state_key) to event_id, for each event that we believe - should be included in the `state` or `state_after` part of the sync response. + should be included in the `state` or `state_after` part of the sync + response, and the subset of its keys whose state changed since + `since_token`. The other keys are memberships fetched for timeline + senders, which the client may have been sent before. """ if members_to_fetch is not None: # Lazy-loading is enabled. Only return the state that is needed. @@ -1549,6 +1648,7 @@ class SyncHandler: # timeline. If at the end we can just use the current state delta stream. if sync_config.use_state_after: delta_state_ids: MutableStateMap[str] = {} + changed_keys: set[StateKey] = set() if members_to_fetch: # We're lazy-loading, so the client might need some more member @@ -1603,9 +1703,11 @@ class SyncHandler: # Note that deltas are in stream ordering, so if there are # multiple deltas for a given type/state_key we'll always pick # the latest one. - delta_state_ids[(delta.event_type, delta.state_key)] = delta.event_id + key = (delta.event_type, delta.state_key) + delta_state_ids[key] = delta.event_id + changed_keys.add(key) - return delta_state_ids + return delta_state_ids, changed_keys # For a non-gappy sync if the events in the timeline are simply a linear # chain (i.e. no merging/branching of the graph), then we know the state @@ -1653,7 +1755,9 @@ class SyncHandler: await_full_state=False, ) ) - return state_ids + # A linear, non-gappy timeline means the only changes to the `state` + # are lazy-loaded members. + return state_ids, set() if batch: state_at_timeline_start = ( @@ -1712,7 +1816,16 @@ class SyncHandler: lazy_load_members=lazy_load_members, ) - return state_ids + # Everything that differs from the state at the previous sync is a + # change. What is left is the memberships of timeline senders that + # `_calculate_state` adds back for lazy loading. + changed_keys = { + key + for key, event_id in state_ids.items() + if state_at_previous_sync.get(key) != event_id + } + + return state_ids, changed_keys async def _find_missing_partial_state_memberships( self, @@ -2650,7 +2763,13 @@ class SyncHandler: """ since_token = sync_result_builder.since_token - user_id = sync_result_builder.sync_config.user.to_string() + sync_config = sync_result_builder.sync_config + user_id = sync_config.user.to_string() + + if sync_config.filter_collection.lazy_load_members(): + self.get_lazy_loaded_members_cache( + (user_id, sync_config.device_id) + ).start_request(since_token) blocks_all_rooms = ( sync_result_builder.sync_config.filter_collection.blocks_all_rooms() @@ -3318,6 +3437,7 @@ class SyncHandler: sync_config, since_token, room_builder.end_token, + now_token, full_state=full_state, joined=room_builder.rtype == "joined", ) diff --git a/tests/handlers/test_sync.py b/tests/handlers/test_sync.py index 2f0107dd0a..1002a0374c 100644 --- a/tests/handlers/test_sync.py +++ b/tests/handlers/test_sync.py @@ -33,7 +33,9 @@ from synapse.api.room_versions import RoomVersion, RoomVersions from synapse.events import EventBase from synapse.events.snapshot import EventContext from synapse.handlers.sync import ( + LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE, LAZY_LOADED_PROFILE_FIELDS_CACHE_MAX_AGE, + LazyLoadedMembersCache, SyncConfig, SyncRequestKey, SyncResult, @@ -48,6 +50,7 @@ from synapse.types import ( MultiWriterStreamToken, RoomStreamToken, StreamKeyType, + StreamToken, UserID, create_requester, ) @@ -2947,7 +2950,7 @@ class SyncStateAfterTestCase(tests.unittest.HomeserverTestCase): # Calculating the incrementals state will return the second state, and not the # first. - state = self.get_success( + state, _ = self.get_success( self.sync_handler._compute_state_delta_for_incremental_sync( room_id=joined_room, sync_config=generate_sync_config(user, use_state_after=True), @@ -2979,7 +2982,7 @@ class SyncStateAfterTestCase(tests.unittest.HomeserverTestCase): since_token = self.hs.get_event_sources().get_current_token() end_stream_token = self.hs.get_event_sources().get_current_token() - state = self.get_success( + state, _ = self.get_success( self.sync_handler._compute_state_delta_for_incremental_sync( room_id=joined_room, sync_config=generate_sync_config(user, use_state_after=True), @@ -2995,3 +2998,76 @@ class SyncStateAfterTestCase(tests.unittest.HomeserverTestCase): ) self.assertEqual(state, {}) + + +class LazyLoadedMembersCacheTestCase(tests.unittest.TestCase): + """Direct tests of the per-device cache of lazy-loaded members sent.""" + + @staticmethod + def _token(stream: int) -> StreamToken: + """A token whose room position is `stream`. Only the room position + matters to the cache.""" + return StreamToken.START.copy_and_replace( + StreamKeyType.ROOM, RoomStreamToken(stream=stream) + ) + + def test_forgets_only_entries_after_since(self) -> None: + """A request from `since` forgets the entries recorded after it and + keeps the rest.""" + cache = LazyLoadedMembersCache() + cache.mark_sent("@a:test", "$a", self._token(5)) + cache.mark_sent("@b:test", "$b", self._token(7)) + + # The client synced past both responses. + cache.start_request(self._token(7)) + self.assertTrue(cache.was_sent("@a:test", "$a")) + self.assertTrue(cache.was_sent("@b:test", "$b")) + + # The client lost the response that ended at 7. + cache.start_request(self._token(6)) + self.assertTrue(cache.was_sent("@a:test", "$a")) + self.assertFalse(cache.was_sent("@b:test", "$b")) + + # A later request from an even older token forgets the rest. + cache.start_request(self._token(4)) + self.assertFalse(cache.was_sent("@a:test", "$a")) + + def test_initial_sync_forgets_everything(self) -> None: + """An initial sync has no `since`, and the client starts from nothing.""" + cache = LazyLoadedMembersCache() + cache.mark_sent("@a:test", "$a", self._token(5)) + cache.start_request(None) + self.assertFalse(cache.was_sent("@a:test", "$a")) + + def test_different_event_is_not_sent(self) -> None: + """An entry only covers the membership event it was recorded with, so + a changed membership counts as not sent.""" + cache = LazyLoadedMembersCache() + cache.mark_sent("@a:test", "$a1", self._token(5)) + self.assertFalse(cache.was_sent("@a:test", "$a2")) + + def test_out_of_order_marks_are_still_forgotten(self) -> None: + """A response that started earlier can record its entries after a + later one. Its entries must still be forgotten by a request from + before either response.""" + cache = LazyLoadedMembersCache() + cache.mark_sent("@b:test", "$b", self._token(7)) + cache.mark_sent("@a:test", "$a", self._token(5)) + + cache.start_request(self._token(6)) + self.assertTrue(cache.was_sent("@a:test", "$a")) + self.assertFalse(cache.was_sent("@b:test", "$b")) + + def test_evicts_least_recently_used(self) -> None: + """The cache holds `LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE` entries and + evicts the least recently used one. A hit counts as a use.""" + cache = LazyLoadedMembersCache() + for i in range(LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE): + cache.mark_sent(f"@u{i}:test", f"$u{i}", self._token(1)) + # Touch the oldest entry so that it is no longer the eviction candidate. + self.assertTrue(cache.was_sent("@u0:test", "$u0")) + + cache.mark_sent("@new:test", "$new", self._token(1)) + self.assertTrue(cache.was_sent("@u0:test", "$u0")) + self.assertFalse(cache.was_sent("@u1:test", "$u1")) + self.assertTrue(cache.was_sent("@new:test", "$new")) diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py index 51ffabe75d..20fdacdc01 100644 --- a/tests/rest/client/test_sync.py +++ b/tests/rest/client/test_sync.py @@ -45,7 +45,7 @@ from tests.federation.transport.test_knocking import ( KnockingStrippedStateEventHelperMixin, ) from tests.rest.client.test_rooms import make_request_with_cancellation_test -from tests.server import TimedOutException +from tests.server import FakeChannel, TimedOutException from tests.test_utils.event_injection import ( inject_event, persist_message_and_state_event_in_one_batch, @@ -1684,3 +1684,334 @@ class SyncStateAfterArchivedRoomTestCase(unittest.HomeserverTestCase): ], [bob_member_event_id_at_leave], ) + + +class SyncLazyLoadedMembersCacheTestCase(unittest.HomeserverTestCase): + """Tests for the per-device cache of lazy-loaded members already sent. + + A membership left out of a response as already sent is lost for good if + the client never processed the response that carried it. The cache must + only trust a response once the client has synced from its `next_batch`. + """ + + servlets = [ + synapse.rest.admin.register_servlets, + room.register_servlets, + login.register_servlets, + sync.register_servlets, + ] + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.alice = self.register_user("alice", "password") + self.alice_device = "alice_device" + self.alice_tok = self.login("alice", "password", device_id=self.alice_device) + self.bob = self.register_user("bob", "password") + self.bob_tok = self.login("bob", "password") + + self.get_success( + hs.get_datastores().main.set_features_for_user( + self.alice, {ExperimentalFeature.MSC4222: True} + ) + ) + + # Named room, so that hero calculation doesn't inject current + # membership state into the response and confuse the assertions. + self.room_id = self.helper.create_room_as( + self.alice, tok=self.alice_tok, extra_content={"name": "test room"} + ) + self.helper.join(self.room_id, self.bob, tok=self.bob_tok) + + # Push bob's join out of the initial sync's timeline, so that his + # membership has not been sent to the client when a test starts. + for i in range(10): + self.helper.send(self.room_id, body=f"pad {i}", tok=self.alice_tok) + + def _sync_request( + self, + use_state_after: bool, + since: str | None = None, + *, + timeline_limit: int = 10, + timeline_not_types: list[str] | None = None, + timeout_ms: int = 0, + await_result: bool = True, + ) -> FakeChannel: + """Make a lazy-loading `/sync` request as alice.""" + timeline_filter: JsonDict = {"limit": timeline_limit} + if timeline_not_types is not None: + timeline_filter["not_types"] = timeline_not_types + sync_filter: JsonDict = { + "room": { + "timeline": timeline_filter, + "state": {"lazy_load_members": True}, + } + } + url = f"/sync?filter={json.dumps(sync_filter)}&timeout={timeout_ms}" + if use_state_after: + url += "&org.matrix.msc4222.use_state_after=true" + if since is not None: + url += f"&since={since}" + return self.make_request( + "GET", url, access_token=self.alice_tok, await_result=await_result + ) + + def _sync( + self, + use_state_after: bool, + since: str | None = None, + *, + timeline_limit: int = 10, + timeline_not_types: list[str] | None = None, + ) -> JsonDict: + """Sync as alice and return the response body.""" + channel = self._sync_request( + use_state_after, + since, + timeline_limit=timeline_limit, + timeline_not_types=timeline_not_types, + ) + self.assertEqual(channel.code, 200, channel.result) + return channel.json_body + + def _room(self, response: JsonDict) -> JsonDict: + """Extract the room dict for the test room from a sync response.""" + rooms = response["rooms"].get("join", {}) + self.assertIn(self.room_id, rooms, f"room missing from sync: {response}") + return rooms[self.room_id] + + def _timeline_ids(self, response: JsonDict) -> list[str]: + return [e["event_id"] for e in self._room(response)["timeline"]["events"]] + + def _state_events( + self, response: JsonDict, use_state_after: bool + ) -> list[JsonDict]: + """Extract the list of state events from the test room in a sync response.""" + key = "org.matrix.msc4222.state_after" if use_state_after else "state" + return self._room(response)[key]["events"] + + def _state_ids(self, response: JsonDict, use_state_after: bool) -> list[str]: + """Extract the list of state event IDs from the test room in a sync + response.""" + return [e["event_id"] for e in self._state_events(response, use_state_after)] + + def _members_in_state(self, response: JsonDict, use_state_after: bool) -> list[str]: + """Extract the list of member state keys from the test room in a sync + response.""" + return [ + e["state_key"] + for e in self._state_events(response, use_state_after) + if e["type"] == EventTypes.Member + ] + + @parameterized.expand([("state", False), ("state_after", True)]) + def test_member_sent_once_when_acknowledged( + self, _: str, use_state_after: bool + ) -> None: + """A lazy-loaded member is sent the first time it is needed and left + out once the client has synced past the response that carried it.""" + t0 = self._sync(use_state_after)["next_batch"] + + self.helper.send(self.room_id, body="one", tok=self.bob_tok) + response = self._sync(use_state_after, t0) + self.assertEqual(self._members_in_state(response, use_state_after), [self.bob]) + t1 = response["next_batch"] + + self.helper.send(self.room_id, body="two", tok=self.bob_tok) + response = self._sync(use_state_after, t1) + self.assertEqual(self._members_in_state(response, use_state_after), []) + + @parameterized.expand([("state", False), ("state_after", True)]) + def test_member_not_resent_on_resync_from_token_that_carried_it( + self, _: str, use_state_after: bool + ) -> None: + """A client that resumes from an older token is only sent again the + members carried by the responses after that token. + + A member sent in the response that ended at the resumed-from token was + processed by the client before it saved that token, so it must not be + sent again. matrix-js-sdk saves its sync store every few minutes and + resumes from the saved token on reload, so this is the common case. + """ + t0 = self._sync(use_state_after)["next_batch"] + + # bob's membership is carried by the response that ends at `t1`. + self.helper.send(self.room_id, body="one", tok=self.bob_tok) + response = self._sync(use_state_after, t0) + self.assertEqual(self._members_in_state(response, use_state_after), [self.bob]) + t1 = response["next_batch"] + + # The client processes further responses, then reloads from `t1`. + self.helper.send(self.room_id, body="two", tok=self.bob_tok) + self._sync(use_state_after, t1) + self.helper.send(self.room_id, body="three", tok=self.bob_tok) + response = self._sync(use_state_after, t1) + self.assertEqual( + self._members_in_state(response, use_state_after), + [], + f"bob's membership re-sent on resume from the token that carried " + f"it: {response}", + ) + + @parameterized.expand([("state", False), ("state_after", True)]) + def test_member_resent_on_retry_from_same_token( + self, _: str, use_state_after: bool + ) -> None: + """A client that gives up on a slow request and retries it must get the + lazy-loaded members again. It never processed the first response.""" + t0 = self._sync(use_state_after)["next_batch"] + + self.helper.send(self.room_id, body="hello", tok=self.bob_tok) + for attempt in range(2): + response = self._sync(use_state_after, t0) + self.assertEqual( + self._members_in_state(response, use_state_after), + [self.bob], + f"bob's membership missing on attempt {attempt}: {response}", + ) + + def test_membership_change_resent_on_resync_from_older_token(self) -> None: + """A membership change must be in `state_after` every time the sync + window covers it, even if it has been sent before. + + Clients do not persist every `since` token. matrix-js-sdk saves its + sync store every few minutes, so after a reload it resumes from a token + older than the one the change was first delivered on. A `state_after` + client never applies state from the timeline, so if the change is left + out of `state_after` it keeps the stale membership until it clears its + cache. A kicked user stays shown as joined. + """ + t0 = self._sync(use_state_after=True)["next_batch"] + + self.helper.send(self.room_id, body="hello", tok=self.bob_tok) + t1 = self._sync(use_state_after=True, since=t0)["next_batch"] + + kick = self.helper.change_membership( + self.room_id, self.alice, self.bob, "leave", tok=self.alice_tok + ) + + # The live client sees the kick. + response = self._sync(use_state_after=True, since=t1) + self.assertIn(kick["event_id"], self._state_ids(response, use_state_after=True)) + + # The client reloads from the older persisted token. + response = self._sync(use_state_after=True, since=t0) + self.assertIn(kick["event_id"], self._timeline_ids(response)) + self.assertIn( + kick["event_id"], + self._state_ids(response, use_state_after=True), + f"membership change missing from state_after on re-sync from an " + f"older token: {response}", + ) + + def test_gappy_membership_change_resent_on_resync_from_older_token( + self, + ) -> None: + """The same for `state`. A membership change that falls in the gap of + a limited timeline is only conveyed by `state`, so it must be there + every time the sync window covers it.""" + t0 = self._sync(use_state_after=False, timeline_limit=1)["next_batch"] + + self.helper.send(self.room_id, body="hello", tok=self.bob_tok) + t1 = self._sync(use_state_after=False, since=t0, timeline_limit=1)["next_batch"] + + kick = self.helper.change_membership( + self.room_id, self.alice, self.bob, "leave", tok=self.alice_tok + ) + # A later message, so that the kick falls into the gap of the + # one-event timeline. + self.helper.send(self.room_id, body="after", tok=self.alice_tok) + + # The live client sees the kick in `state`. + response = self._sync(use_state_after=False, since=t1, timeline_limit=1) + self.assertTrue(self._room(response)["timeline"]["limited"]) + self.assertNotIn(kick["event_id"], self._timeline_ids(response)) + self.assertIn( + kick["event_id"], self._state_ids(response, use_state_after=False) + ) + + # The client reloads from the older persisted token. + response = self._sync(use_state_after=False, since=t0, timeline_limit=1) + self.assertNotIn(kick["event_id"], self._timeline_ids(response)) + self.assertIn( + kick["event_id"], + self._state_ids(response, use_state_after=False), + f"membership change missing from state on re-sync from an older " + f"token: {response}", + ) + + @parameterized.expand([("state", False), ("state_after", True)]) + def test_member_not_resent_after_empty_long_poll_wakeup( + self, _: str, use_state_after: bool + ) -> None: + """A long-poll that wakes up with nothing to send keeps waiting. The + response it eventually returns must still leave out the members the + client already has. + + The notifier calls back into the sync handler on every wakeup within + one request. An empty result is not a response the client lost. + """ + not_types = ["m.spurious"] + t0 = self._sync(use_state_after, timeline_not_types=not_types)["next_batch"] + + self.helper.send(self.room_id, body="one", tok=self.bob_tok) + response = self._sync(use_state_after, t0, timeline_not_types=not_types) + self.assertEqual(self._members_in_state(response, use_state_after), [self.bob]) + t1 = response["next_batch"] + + channel = self._sync_request( + use_state_after, + t1, + timeline_not_types=not_types, + timeout_ms=30000, + await_result=False, + ) + + # Wakes the long-poll, but the filter drops the event, so the result is + # empty and the notifier waits again. + self.helper.send_event( + self.room_id, type="m.spurious", content={}, tok=self.bob_tok + ) + self.assertFalse(channel.is_finished()) + + self.helper.send(self.room_id, body="two", tok=self.bob_tok) + channel.await_result() + self.assertEqual(channel.code, 200, channel.result) + response = channel.json_body + self.assertEqual(len(self._timeline_ids(response)), 1) + self.assertEqual(self._members_in_state(response, use_state_after), []) + + @parameterized.expand([("state", False), ("state_after", True)]) + def test_membership_change_sent_when_cache_claims_it_was( + self, _: str, use_state_after: bool + ) -> None: + """A membership change since `since` is sent even when the cache says + the client already has it. + + The cache only decides whether to leave out the memberships fetched for + timeline senders. It can be wrong: a retry of a request can run while + the original is still being built, and the original then marks members + the retry has to send. A change must never depend on it. + """ + t0 = self._sync(use_state_after, timeline_limit=1)["next_batch"] + + kick = self.helper.change_membership( + self.room_id, self.alice, self.bob, "leave", tok=self.alice_tok + ) + # A later message, so that the kick falls into the gap of the + # one-event timeline and only `state` can carry it. + self.helper.send(self.room_id, body="after the kick", tok=self.alice_tok) + + # Claim the kick was already sent in the response that ended at `t0`. + # Syncing from `t0` does not forget this, as the client has acknowledged + # that response. + t0_token = self.get_success( + StreamToken.from_string(self.hs.get_datastores().main, t0) + ) + cache = self.hs.get_sync_handler().get_lazy_loaded_members_cache( + (self.alice, self.alice_device) + ) + cache.mark_sent(self.bob, kick["event_id"], t0_token) + + response = self._sync(use_state_after, t0, timeline_limit=1) + self.assertNotIn(kick["event_id"], self._timeline_ids(response)) + self.assertIn(kick["event_id"], self._state_ids(response, use_state_after))