diff --git a/changelog.d/20433.bugfix b/changelog.d/20433.bugfix new file mode 100644 index 0000000000..5c2fa2f0f5 --- /dev/null +++ b/changelog.d/20433.bugfix @@ -0,0 +1 @@ +Fix race where an initial sync did not return some memberships in rooms when another sync for the device was in-flight. Introduced in v1.163.0. diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index d635465e5c..c35e8f01fb 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -1437,19 +1437,23 @@ class SyncHandler: # 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. + # cache). # # 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) + # + # We don't filter the full state as we know this is the first + # time the client will see the memberships. For full state the + # cache *should* already be cleared by this point. + if not full_state: + 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 cache for t, event_id in itertools.chain( diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py index bf00c0f331..4f80758033 100644 --- a/tests/rest/client/test_sync.py +++ b/tests/rest/client/test_sync.py @@ -2138,3 +2138,52 @@ class SyncLazyLoadedMembersCacheTestCase(unittest.HomeserverTestCase): 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)) + + @parameterized.expand([("state", False), ("state_after", True)]) + def test_full_state_is_not_filtered_through_cache( + self, _: str, use_state_after: bool + ) -> None: + """A newly joined room is sent its full (lazy-loaded) state, including + the user's own membership, even if the cache claims that membership was + already sent. + + This test should only be needed while there are gaps in the lazy-loaded + membership cache logic. Specifically, around its handling of concurrent + sync requests. + """ + + # Do a sync before joining a new room. + t0 = self._sync(use_state_after)["next_batch"] + + room2 = self.helper.create_room_as( + self.bob, tok=self.bob_tok, extra_content={"name": "room 2"} + ) + self.helper.invite(room2, self.bob, self.alice, tok=self.bob_tok) + join = self.helper.join(room2, self.alice, tok=self.alice_tok) + + # Push the join out of the 10 event timeline, so that only the state + # section can carry alice's membership. + for i in range(10): + self.helper.send(room2, body=f"welcome {i}", tok=self.bob_tok) + + # Directly change the cache so that it thinks the join was already sent. + 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.alice, join["event_id"], t0_token) + + # Do a sync from t0, which will include the full state for the newly + # joined room. Despite what the cache says, the full state should still + # include the user's own membership. + response = self._sync(use_state_after, t0) + + key = "state_after" if use_state_after else "state" + members = [ + e["state_key"] + for e in response["rooms"]["join"][room2][key]["events"] + if e["type"] == EventTypes.Member + ] + self.assertIn(self.alice, members)