Fix bug where /sync would not return membership changes due to lazy-loaded membership cache (#20289)

In `/sync` there is a mode called lazy loaded (LL) members. In this mode
`/sync` does not return all memberships in the room, instead it only
returns a) membership changes, and b) the memberships of senders of
events in the timeline (if not previously sent).

Synapse implements a lazy-loaded member cache to avoid repeatedly
sending down the same memberships when a single user sends multiple
events. It does this by storing for each user/device the memberships it
has previously returned in an LRU cache. Note that it is fine for the
server to return the same membership, clients handle that fine, but it
is a waste of bandwidth.

The current lazy-loaded membership cache has three intertwined flaws:
1. It does not handle retries or clients re-using old tokens.
2. It incorrectly applies the membership cache to membership changes,
not just memberships added by lazy-loading.
3. It does not correctly handle multiple "sync streams" from the same
device.

In this PR we explicitly fix the first two issues in two commits, and
the third issue is mitigated:

1. Alongside each membership in the LL cache we store the `next_batch`
position of the response that carried it. A client that processed a
response syncs from its `next_batch` next. When a request arrives with a
`since` token, we delete every membership recorded after that token, as
they were sent in responses the client never processed. Memberships
recorded at or before `since` are kept.
2. We always return memberships that have changed, even if they appear
in the LL member cache.

These changes also half mitigate the issue of running multiple `/sync`
loops per-device (though we do not see this in the wild), as we'd at
least always return the changed membership states.

Fixes #19978 and #20278, supersedes #20279

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`.

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",
                 )

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"))

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))