Fix regression with lazy membership and concurrenet syncs (#20433)

Broke by #20289. Hopefully fixes element-hq/element-web#35567.

One subtle change in #20289 was that we started to filter rooms with
`full_state` set through the cache. This *should* be safe since none of
the membership events should have been previously sent down the sync.

However, there is a bug with the lazy-loaded member cache when two
concurrent syncs for the same device happens. The updates to the cache
from the handling of one request are immediately seen by the other
request, polluting the cache.

This is a relatively common scenario in Element Web when a user reloads
the page. The sync from before the reload is still being processed when
Synapse receives the initial sync from after the reload.

The full fix to correctly handle overlapping requests is a separate PR,
but this is a stop-gap measure that should help Element Web.

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.

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(

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)