Stage the lazy-loaded members a sync response sends until it finishes alone

Two `/sync` responses for one device can be built at the same time: a
retry of a request that is still running, a reload while the previous
request is still running, or a client with two sync loops. Whatever
processes one of them has not seen the other. `LazyLoadedMembersCache`
forgets the entries recorded after `since` at the start of a request,
but the other response records its entries after that, and the request
then leaves out members the client never received.

This is the race behind element-hq/element-web#35567. #20433 stopped
filtering full state through the cache, which keeps the user's own
membership safe. The incremental path still loses the memberships of
timeline senders the same way: a reload from an older persisted token
that overlaps the request from before the reload leaves out whatever
that request recorded, and a `state_after` client never recovers them.

A response now stages what it sends instead of recording it as it goes.
`generate_sync_result` builds the response inside
`cache.building_response(since_token)`, which forgets the entries the
client did not receive and yields a `LazyLoadedMembersHandle`. The
handle reads the cache and stages the memberships the response sends.
The cache keeps the handles of the responses being built; a new response
clears the `commits` flag on itself and on every one of them. When a
response finishes, the cache commits its staged entries if the flag is
still set. Responses that overlapped commit nothing and their members
are sent again. A response that raised, which includes a cancelled
request, commits nothing either. A healthy client has one request in
flight at a time, so nothing changes for it.

A response that had to forget entries does not commit either. Two
clients sharing a device build responses one after another and may end
at the same position, and an entry one of them records at that position
looks acknowledged to the other. The next response from the first one's
`next_batch` finds nothing to forget and commits again. This costs one
more re-send after a retry or a reload.

The cache is looked up once per request and the handle carried on the
`SyncResultBuilder`, so the `ExpiringCache` may still drop the cache
after 30 minutes: an object replaced under a running request only serves
that request.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SXqkiTLLbfxpYt9dtAEkeQ

changelog.d/20435.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

@@ -24,11 +24,13 @@ import json
 import logging
 import os
 from collections import OrderedDict
+from contextlib import contextmanager
 from typing import (
     TYPE_CHECKING,
     AbstractSet,
     Any,
     Collection,
+    Iterator,
     Mapping,
     Sequence,
 )
@@ -138,17 +140,22 @@ class LazyLoadedMembersCache:
     """The membership events a lazy-loading client has been sent, so that later
     syncs can leave them out. One per (user, device).
 
+    The cache needs to handle concurrent requests, retries, and reusing old
+    tokens. Ideally, the sync token would include a field that can be used to
+    track the state of the cache, similarly to sliding sync. The following
+    details how the cache handles the various scenarios without an ID.
+
     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.
+    processed every response up to `since` and none after it.
+    `building_response` 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
@@ -162,33 +169,93 @@ class LazyLoadedMembersCache:
     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.
+
+    The above handles reusing old tokens. However, we need to also guard against
+    concurrent requests and retries from the same device. A typical example is
+    when Element Web gets reloaded, and Synapse processes an incremental sync
+    request from before the reload at the same time as the initial sync request
+    triggered by the reload.
+
+    We solve this by detecting when a concurrent request has happened and
+    invalidating/not committing any entries from *any* concurrent request. This
+    is because we don't know which request the client will actually process.
+
+    There are two distinct scenarios:
+      1. The response for the two requests are processed at the same time.
+      2. One or other finishes processing before the other starts.
+
+    For the first scenario, we track how many in-flight responses are being
+    built for this device. If a second response starts building while another is
+    still in progress, it will ensure that neither response commits any entries
+    to the cache.
+
+    For the second scenario, the second response notices that the are entries in
+    the cache that it needs to evict. It therefore invalidates those and doesn't
+    commit its own entries.
     """
 
     # 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`
+    # Not an `LruCache`, as that cannot be iterated over and `_forget_after`
     # 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`
+    # An upper bound on the positions of the entries, so that `_forget_after`
     # 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."""
+    # The responses being built for this device right now.
+    _building: list["LazyLoadedMembersHandle"] = attr.Factory(list)
+
+    @contextmanager
+    def building_response(
+        self, since_token: StreamToken | None
+    ) -> Iterator["LazyLoadedMembersHandle"]:
+        """Wraps building one response for this device, syncing from
+        `since_token`. Yields the handle the response reads and stages
+        through. Once the response has finished, commits what it staged,
+        unless its `commits` flag was cleared (see the class docstring)."""
+        handle = LazyLoadedMembersHandle(cache=self)
+
+        # Check if there are multiple responses being processed at once, if so
+        # none of them may commit.
+        if self._building:
+            handle.commits = False
+            for other in self._building:
+                other.commits = False
+
+        # If any entries get invalidated, it indicates that there are concurrent
+        # requests and we should not commit.
+        if self._forget_after(since_token):
+            handle.commits = False
+
+        self._building.append(handle)
+        try:
+            yield handle
+        finally:
+            self._building.remove(handle)
+
+        # Not reached if building the response failed, which includes the
+        # request being cancelled.
+        if handle.commits:
+            for user_id, (event_id, sent_at) in handle.staged.items():
+                self.record(user_id, event_id, sent_at)
+
+    def _forget_after(self, since_token: StreamToken | None) -> bool:
+        """Forget the members the client did not receive: the entries recorded
+        after `since_token`. Returns whether there were any."""
         if since_token is None:
             self._sent.clear()
             self._newest = None
-            return
+            return False
 
         # 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
+            return False
 
         # Remove all entries that were sent after `since`. This only has
         # `LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE` entries.
@@ -199,6 +266,11 @@ class LazyLoadedMembersCache:
         # Everything left is at or before `since`.
         self._newest = since
 
+        # Report by the bound, not by whether anything was deleted. The
+        # entries may all have been evicted already, and the responses that
+        # carried them still ended after `since`.
+        return True
+
     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)
@@ -207,22 +279,48 @@ class LazyLoadedMembersCache:
         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
+    def record(self, user_id: str, event_id: str, sent_at: RoomStreamToken) -> None:
+        """Record that `event_id` is the membership of `user_id` the client
+        received, in the response ending at room position `sent_at`."""
         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.
+        # Responses commit one after another and positions only advance, so
+        # `sent_at` is normally the newest. Never move the bound back
+        # regardless, as `_forget_after` would then skip the entries above it.
         if self._newest is None or self._newest.is_before_or_eq(sent_at):
             self._newest = sent_at
 
 
+@attr.s(slots=True, auto_attribs=True)
+class LazyLoadedMembersHandle:
+    """One response's view of its device's `LazyLoadedMembersCache`. Reads go
+    to the cache. What the response sends is staged here, and the cache
+    commits it when the response finishes if `commits` is still set. See
+    `building_response`."""
+
+    _cache: LazyLoadedMembersCache
+
+    # Whether the cache commits what this response staged when it finishes.
+    # Cleared when another response is being built at any point while this
+    # one is, or when this one had to forget entries.
+    commits: bool = True
+
+    # user_id -> (event_id, room position of `next_batch`), as in the cache.
+    staged: dict[str, tuple[str, RoomStreamToken]] = attr.Factory(dict)
+
+    def was_sent(self, user_id: str, event_id: str) -> bool:
+        """Whether `event_id` is the membership of `user_id` we last sent."""
+        return self._cache.was_sent(user_id, event_id)
+
+    def mark_sent(self, user_id: str, event_id: str, next_batch: StreamToken) -> None:
+        """Stage that `event_id` is being sent in the response ending at
+        `next_batch`."""
+        self.staged[user_id] = (event_id, next_batch.room_key)
+
+
 SyncRequestKey = tuple[Any, ...]
 
 
@@ -1107,12 +1205,16 @@ class SyncHandler:
         batch: TimelineBatch,
         state: MutableStateMap[EventBase],
         now_token: StreamToken,
+        lazy_loaded_members: LazyLoadedMembersHandle | None,
     ) -> JsonDict | None:
         """Works out a room summary block for this room, summarising the number
         of joined members in the room, and providing the 'hero' members if the
         room has no name so clients can consistently name rooms.  Also adds
         state events to 'state' if needed to describe the heroes.
 
+        `lazy_loaded_members` is the response's `LazyLoadedMembersHandle`, or
+        None if the request does not lazy-load members.
+
         Args
             room_id
             sync_config
@@ -1187,8 +1289,8 @@ class SyncHandler:
             return summary
 
         # ensure we send membership events for heroes if needed
-        cache_key = (sync_config.user.to_string(), sync_config.device_id)
-        cache = self.get_lazy_loaded_members_cache(cache_key)
+        assert lazy_loaded_members is not None
+        cache = lazy_loaded_members
 
         # track which members the client should already know about via LL:
         # Ones which are already in state...
@@ -1271,6 +1373,7 @@ class SyncHandler:
         now_token: StreamToken,
         full_state: bool,
         joined: bool,
+        lazy_loaded_members: LazyLoadedMembersHandle | None,
     ) -> MutableStateMap[EventBase]:
         """Works out the difference in state between the end of the previous sync and
         the start of the timeline.
@@ -1287,6 +1390,8 @@ class SyncHandler:
             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
+            lazy_loaded_members: The response's `LazyLoadedMembersHandle`, or
+                None if the request does not lazy-load members.
 
         Returns:
             The state to return in the sync response for the room.
@@ -1432,8 +1537,8 @@ class SyncHandler:
             # memberships of event senders. Otherwise we make an effort to limit the set
             # of memberships we send to those that we have not already sent to this client.
             if lazy_load_members and not include_redundant_members:
-                cache_key = (sync_config.user.to_string(), sync_config.device_id)
-                cache = self.get_lazy_loaded_members_cache(cache_key)
+                assert lazy_loaded_members is not None
+                cache = lazy_loaded_members
 
                 # Only send members which aren't in our cache (either because
                 # they're new to this client or have been pushed out of the
@@ -2063,6 +2168,33 @@ class SyncHandler:
         sync_config: SyncConfig,
         since_token: StreamToken | None = None,
         full_state: bool = False,
+    ) -> SyncResult:
+        """Generates the response body of a sync result. See
+        `_generate_sync_result`.
+
+        Tells the device's `LazyLoadedMembersCache` when the response starts
+        and finishes being built. That is how it detects two responses being
+        built at once.
+        """
+        if not sync_config.filter_collection.lazy_load_members():
+            return await self._generate_sync_result(
+                sync_config, since_token, full_state, lazy_loaded_members=None
+            )
+
+        cache = self.get_lazy_loaded_members_cache(
+            (sync_config.user.to_string(), sync_config.device_id)
+        )
+        with cache.building_response(since_token) as handle:
+            return await self._generate_sync_result(
+                sync_config, since_token, full_state, lazy_loaded_members=handle
+            )
+
+    async def _generate_sync_result(
+        self,
+        sync_config: SyncConfig,
+        since_token: StreamToken | None,
+        full_state: bool,
+        lazy_loaded_members: LazyLoadedMembersHandle | None,
     ) -> SyncResult:
         """Generates the response body of a sync result.
 
@@ -2089,6 +2221,7 @@ class SyncHandler:
             since_token,
             full_state,
         )
+        sync_result_builder.lazy_loaded_members = lazy_loaded_members
 
         logger.debug(
             "Calculating sync response for %r between %s and %s",
@@ -2871,11 +3004,6 @@ class SyncHandler:
         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()
         )
@@ -3569,6 +3697,7 @@ class SyncHandler:
                     now_token,
                     full_state=full_state,
                     joined=room_builder.rtype == "joined",
+                    lazy_loaded_members=sync_result_builder.lazy_loaded_members,
                 )
             else:
                 # An out of band room won't have any state changes.
@@ -3598,7 +3727,12 @@ class SyncHandler:
                 )
             ):
                 summary = await self.compute_summary(
-                    room_id, sync_config, batch, state, now_token
+                    room_id,
+                    sync_config,
+                    batch,
+                    state,
+                    now_token,
+                    sync_result_builder.lazy_loaded_members,
                 )
 
             if room_builder.rtype == "joined":
@@ -3812,6 +3946,8 @@ class SyncResultBuilder:
             (This is useful if the room was previously excluded from a /sync response,
             and now the client should be made aware of it.)
             Only used by incremental syncs.
+        lazy_loaded_members: The response's `LazyLoadedMembersHandle`, or None
+            if the request does not lazy-load members.
 
         # The following mirror the fields in a sync response
         presence
@@ -3833,6 +3969,8 @@ class SyncResultBuilder:
     forced_newly_joined_room_ids: frozenset[str]
     membership_change_events: list[EventBase]
 
+    lazy_loaded_members: LazyLoadedMembersHandle | None = None
+
     presence: list[UserPresenceState] = attr.Factory(list)
     account_data: list[JsonDict] = attr.Factory(list)
     profile_updates: dict[str, dict[str, JsonValue | dict[str, JsonValue]] | None] = (

tests/handlers/test_sync.py

@@ -3331,35 +3331,39 @@ class LazyLoadedMembersCacheTestCase(tests.unittest.TestCase):
         """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))
+        cache.record("@a:test", "$a", RoomStreamToken(stream=5))
+        cache.record("@b:test", "$b", RoomStreamToken(stream=7))
 
         # The client synced past both responses.
-        cache.start_request(self._token(7))
+        with cache.building_response(self._token(7)):
+            pass
         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))
+        with cache.building_response(self._token(6)):
+            pass
         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))
+        with cache.building_response(self._token(4)):
+            pass
         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)
+        cache.record("@a:test", "$a", RoomStreamToken(stream=5))
+        with cache.building_response(None):
+            pass
         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))
+        cache.record("@a:test", "$a1", RoomStreamToken(stream=5))
         self.assertFalse(cache.was_sent("@a:test", "$a2"))
 
     def test_out_of_order_marks_are_still_forgotten(self) -> None:
@@ -3367,10 +3371,11 @@ class LazyLoadedMembersCacheTestCase(tests.unittest.TestCase):
         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.record("@b:test", "$b", RoomStreamToken(stream=7))
+        cache.record("@a:test", "$a", RoomStreamToken(stream=5))
 
-        cache.start_request(self._token(6))
+        with cache.building_response(self._token(6)):
+            pass
         self.assertTrue(cache.was_sent("@a:test", "$a"))
         self.assertFalse(cache.was_sent("@b:test", "$b"))
 
@@ -3379,11 +3384,94 @@ class LazyLoadedMembersCacheTestCase(tests.unittest.TestCase):
         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))
+            cache.record(f"@u{i}:test", f"$u{i}", RoomStreamToken(stream=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))
+        cache.record("@new:test", "$new", RoomStreamToken(stream=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"))
+
+    def test_forgetting_disables_recording(self) -> None:
+        """A response from before the newest entry forgets the newer entries
+        and commits nothing itself. A response from its `next_batch` commits
+        again."""
+        cache = LazyLoadedMembersCache()
+        cache.record("@a:test", "$a", RoomStreamToken(stream=5))
+
+        with cache.building_response(self._token(4)) as handle:
+            self.assertFalse(handle.commits)
+            self.assertFalse(handle.was_sent("@a:test", "$a"))
+            handle.mark_sent("@b:test", "$b", self._token(6))
+        self.assertFalse(cache.was_sent("@b:test", "$b"))
+
+        with cache.building_response(self._token(6)) as handle:
+            self.assertTrue(handle.commits)
+            handle.mark_sent("@c:test", "$c", self._token(7))
+        self.assertTrue(cache.was_sent("@c:test", "$c"))
+
+    def test_eviction_still_disables_recording(self) -> None:
+        """The newest-entry bound decides, not the entries. An initial sync
+        can record more members than the cache holds, and the evicted ones
+        still ended after a response's `since`."""
+        cache = LazyLoadedMembersCache()
+        cache.record("@old:test", "$old", RoomStreamToken(stream=5))
+        for i in range(LAZY_LOADED_MEMBERS_CACHE_MAX_SIZE):
+            cache.record(f"@u{i}:test", f"$u{i}", RoomStreamToken(stream=5))
+        self.assertFalse(cache.was_sent("@old:test", "$old"))
+
+        with cache.building_response(self._token(4)) as handle:
+            self.assertFalse(handle.commits)
+
+    def test_overlapping_responses_commit_nothing(self) -> None:
+        """Responses being built at the same time cannot see each other's
+        entries and commit none of their own, whichever finishes first."""
+        cache = LazyLoadedMembersCache()
+        cache.record("@old:test", "$old", RoomStreamToken(stream=4))
+
+        with cache.building_response(self._token(4)) as first:
+            self.assertTrue(first.commits)
+            first.mark_sent("@a:test", "$a", self._token(5))
+
+            with cache.building_response(self._token(4)) as second:
+                self.assertFalse(first.commits)
+                self.assertFalse(second.commits)
+                # Entries committed before the overlap are still used.
+                self.assertTrue(second.was_sent("@old:test", "$old"))
+                self.assertFalse(second.was_sent("@a:test", "$a"))
+                second.mark_sent("@b:test", "$b", self._token(5))
+
+                # A third response overlaps too, and finishes before them.
+                with cache.building_response(self._token(4)) as third:
+                    self.assertFalse(third.commits)
+                    third.mark_sent("@c:test", "$c", self._token(5))
+                self.assertFalse(cache.was_sent("@c:test", "$c"))
+
+            self.assertFalse(cache.was_sent("@b:test", "$b"))
+            first.mark_sent("@d:test", "$d", self._token(5))
+
+        # The first was alone when it started, but not throughout.
+        self.assertFalse(cache.was_sent("@a:test", "$a"))
+        self.assertFalse(cache.was_sent("@d:test", "$d"))
+        self.assertTrue(cache.was_sent("@old:test", "$old"))
+
+        # The next response is alone again and commits.
+        with cache.building_response(self._token(5)) as handle:
+            self.assertTrue(handle.commits)
+            handle.mark_sent("@e:test", "$e", self._token(6))
+        self.assertTrue(cache.was_sent("@e:test", "$e"))
+
+    def test_failed_response_commits_nothing(self) -> None:
+        """A response that was not finished never reached the client."""
+        cache = LazyLoadedMembersCache()
+        with self.assertRaises(RuntimeError):
+            with cache.building_response(None) as handle:
+                handle.mark_sent("@a:test", "$a", self._token(5))
+                if handle.commits:
+                    raise RuntimeError("cancelled")
+        self.assertFalse(cache.was_sent("@a:test", "$a"))
+
+        # And the cache is not left thinking the response is still being built.
+        with cache.building_response(self._token(5)) as handle:
+            self.assertTrue(handle.commits)

tests/rest/client/test_sync.py

@@ -23,6 +23,7 @@ import logging
 
 from parameterized import parameterized
 
+from twisted.internet.defer import Deferred
 from twisted.internet.testing import MemoryReactor
 
 import synapse.rest.admin
@@ -37,7 +38,14 @@ from synapse.api.constants import (
 from synapse.events import EventBase
 from synapse.rest.client import devices, knock, login, read_marker, receipts, room, sync
 from synapse.server import HomeServer
-from synapse.types import JsonDict, RoomStreamToken, StreamKeyType, StreamToken
+from synapse.types import (
+    JsonDict,
+    RoomStreamToken,
+    StateKey,
+    StateMap,
+    StreamKeyType,
+    StreamToken,
+)
 from synapse.util.clock import Clock
 
 from tests import unittest
@@ -1926,6 +1934,27 @@ class SyncLazyLoadedMembersCacheTestCase(unittest.HomeserverTestCase):
             if e["type"] == EventTypes.Member
         ]
 
+    def _park_incremental_path(self) -> tuple[Deferred, list[str]]:
+        """Make the next incremental room state computation wait on the
+        returned Deferred. By then the request has entered
+        `building_response`. The list receives the room ID of the parked
+        computation."""
+        handler = self.hs.get_sync_handler()
+        original = handler._compute_state_delta_for_incremental_sync
+        gate: "Deferred[None]" = Deferred()
+        parked: list[str] = []
+
+        async def wrapper(
+            room_id: str, *args: object, **kwargs: object
+        ) -> tuple[StateMap[str], set[StateKey]]:
+            if not parked:
+                parked.append(room_id)
+                await gate
+            return await original(room_id, *args, **kwargs)  # type: ignore[arg-type]
+
+        handler._compute_state_delta_for_incremental_sync = wrapper  # type: ignore[method-assign]
+        return gate, parked
+
     @parameterized.expand([("state", False), ("state_after", True)])
     def test_member_sent_once_when_acknowledged(
         self, _: str, use_state_after: bool
@@ -2133,7 +2162,7 @@ class SyncLazyLoadedMembersCacheTestCase(unittest.HomeserverTestCase):
         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)
+        cache.record(self.bob, kick["event_id"], t0_token.room_key)
 
         response = self._sync(use_state_after, t0, timeline_limit=1)
         self.assertNotIn(kick["event_id"], self._timeline_ids(response))
@@ -2173,7 +2202,7 @@ class SyncLazyLoadedMembersCacheTestCase(unittest.HomeserverTestCase):
         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)
+        cache.record(self.alice, join["event_id"], t0_token.room_key)
 
         # 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
@@ -2187,3 +2216,71 @@ class SyncLazyLoadedMembersCacheTestCase(unittest.HomeserverTestCase):
             if e["type"] == EventTypes.Member
         ]
         self.assertIn(self.alice, members)
+
+    @parameterized.expand([("state", False), ("state_after", True)])
+    def test_reload_overlapping_old_request(
+        self, _: str, use_state_after: bool
+    ) -> None:
+        """An incremental sync built while another request for the same
+        device is still running must send the members that request sends too
+        (element-hq/element-web#35567).
+
+        Element Web reloads from a token persisted minutes earlier, while the
+        request from before the reload may still be building on the server.
+        That request finishes after the reload's request forgot its entries,
+        and the reload's request must not leave out what it recorded.
+        """
+        t0 = self._sync(use_state_after)["next_batch"]
+        # bob's membership is sent and recorded in the response ending at
+        # t_live, which the reload will not have processed.
+        self.helper.send(self.room_id, body="pad", tok=self.bob_tok)
+        t_live = self._sync(use_state_after, t0)["next_batch"]
+        self.helper.send(self.room_id, body="hi", tok=self.bob_tok)
+
+        gate, parked = self._park_incremental_path()
+        reload = self._sync_request(use_state_after, t0, await_result=False)
+        self.pump()
+        self.assertEqual(parked, [self.room_id])
+
+        # The request from before the reload sends bob's membership again,
+        # as the reload's request forgot it.
+        old = self._sync(use_state_after, t_live)
+        self.assertEqual(self._members_in_state(old, use_state_after), [self.bob])
+
+        gate.callback(None)
+        reload.await_result()
+        self.assertEqual(reload.code, 200, reload.result)
+        self.assertEqual(
+            self._members_in_state(reload.json_body, use_state_after), [self.bob]
+        )
+
+    @parameterized.expand([("state", False), ("state_after", True)])
+    def test_two_clients_on_one_device_do_not_trust_each_other(
+        self, _: str, use_state_after: bool
+    ) -> None:
+        """Two clients sharing a device build responses one after another and
+        end at the same position. What one of them was sent must not be left
+        out for the other.
+
+        Client X syncs from before charlie's join. Client Y starts with an
+        initial sync whose timeline no longer holds the join, so Y has never
+        seen charlie's membership. X's next response carries the join and
+        ends at the same position Y's did. When charlie then sends a message,
+        Y must still be sent his membership.
+        """
+        charlie = self.register_user("charlie", "password")
+        charlie_tok = self.login("charlie", "password")
+
+        x0 = self._sync(use_state_after)["next_batch"]
+
+        self.helper.join(self.room_id, charlie, tok=charlie_tok)
+        for i in range(10):
+            self.helper.send(self.room_id, body=f"after join {i}", tok=self.alice_tok)
+
+        y0 = self._sync(use_state_after)["next_batch"]
+        x1 = self._sync(use_state_after, x0)["next_batch"]
+        self.assertEqual(x1, y0)
+
+        self.helper.send(self.room_id, body="hello", tok=charlie_tok)
+        response = self._sync(use_state_after, y0)
+        self.assertIn(charlie, self._members_in_state(response, use_state_after))