From 6333891e234e1e8b1b59414080a0ee17852c050e Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Fri, 28 Aug 2026 18:30:30 +0200 Subject: [PATCH 1/5] MSC4222: add tests for state events missing from state_after A since token that falls inside a persist batch drops the deltas of the batch's state events, because current_state_delta_stream rows are stamped with the batch minimum stream ordering while the timeline is selected on the event's own ordering. A worker reading the events stream from replication routinely observes such a token, so on worker deployments a state event can be served in the sync timeline of a room the user is still joined to while being absent from state_after, breaking the MSC4222 invariant that `state at since` + `state_after` equals the state at the end of the timeline. The tests document the storage-level and replication-level preconditions, reproduce the bug end-to-end (including the gappy variant where the affected state event is truncated out of the timeline), and guard that state_after reports the resolved state at the end of the timeline rather than a replay of the timeline. Three tests fail at this commit; the next commit fixes them. --- tests/rest/client/test_sync.py | 415 ++++++++++++++++++++++++++++++++- 1 file changed, 414 insertions(+), 1 deletion(-) diff --git a/tests/rest/client/test_sync.py b/tests/rest/client/test_sync.py index 74a8678ae99..ef3d84c2ca0 100644 --- a/tests/rest/client/test_sync.py +++ b/tests/rest/client/test_sync.py @@ -33,9 +33,13 @@ ReceiptTypes, RelationTypes, ) +from synapse.events import EventBase +from synapse.replication.tcp.resource import _batch_updates +from synapse.replication.tcp.streams.events import EventsStream +from synapse.rest.admin.experimental_features import ExperimentalFeature from synapse.rest.client import devices, knock, login, read_marker, receipts, room, sync from synapse.server import HomeServer -from synapse.types import JsonDict +from synapse.types import JsonDict, RoomStreamToken, StreamKeyType, StreamToken from synapse.util.clock import Clock from tests import unittest @@ -44,6 +48,7 @@ ) from tests.rest.client.test_rooms import make_request_with_cancellation_test from tests.server import TimedOutException +from tests.test_utils.event_injection import create_event, inject_event logger = logging.getLogger(__name__) @@ -1272,3 +1277,411 @@ def test_incremental_sync(self) -> None: ) self.assertEqual(200, channel.code, msg=channel.result["body"]) + + +class SyncStateAfterTimelineStateTestCase(unittest.HomeserverTestCase): + """Tests for the MSC4222 invariant that any state event served in a sync + response's *timeline* is reflected in `state_after` (unless it lost state + resolution): `state at since` + `state_after` must equal the state at the + end of the timeline. See + https://github.com/matrix-org/matrix-spec-proposals/pull/4222. + + This breaks when the `since` token falls inside a persist batch: the + `current_state_delta_stream` row for a state event persisted in a batch is + stamped with the *minimum* stream ordering of the batch, so such a token + selects the event for the timeline but not its delta. A worker reading the + events stream from replication routinely observes such a token, which is + why this is seen on worker deployments and not on a single process. + + The remaining tests guard that whatever repairs this reports the *resolved* + state at the end of the timeline rather than replaying the timeline. + + The first two tests are storage- and replication-level and would normally + live under `tests/storage/` / `tests/replication/`; they are kept here + deliberately, as the documented preconditions of the end-to-end tests + beside them (with which they share the `_persist_batch` helper). + """ + + 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.store = hs.get_datastores().main + self.persistence = hs.get_storage_controllers().persistence + assert self.persistence is not None + + self.alice = self.register_user("alice", "password") + self.alice_tok = self.login("alice", "password") + self.bob = self.register_user("bob", "password") + self.bob_tok = self.login("bob", "password") + + self.get_success( + self.store.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) + + # ------------------------------------------------------------------ + # helpers + # ------------------------------------------------------------------ + + def _sync_url(self, lazy_load_members: bool, timeline_limit: int = 10) -> str: + # The default `timeline_limit` of 10 is arbitrary: comfortably more + # events than any test here produces between syncs, so the timeline is + # never truncated unless a test passes a smaller limit on purpose. + sync_filter: JsonDict = {"room": {"timeline": {"limit": timeline_limit}}} + if lazy_load_members: + sync_filter["room"]["state"] = {"lazy_load_members": True} + return f"/sync?filter={json.dumps(sync_filter)}&org.matrix.msc4222.use_state_after=true" + + def _sync(self, sync_url: str, since: str | None = None) -> JsonDict: + url = sync_url if since is None else f"{sync_url}&since={since}" + channel = self.make_request("GET", url, access_token=self.alice_tok) + self.assertEqual(channel.code, 200, channel.result) + return channel.json_body + + def _joined_room(self, response: JsonDict) -> JsonDict: + 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, room: JsonDict) -> list[str]: + return [e["event_id"] for e in room["timeline"]["events"]] + + def _state_after_ids(self, room: JsonDict) -> list[str]: + return [e["event_id"] for e in room["org.matrix.msc4222.state_after"]["events"]] + + def _persist_batch(self) -> tuple[EventBase, EventBase]: + """Persist a message and a state event in a *single* persist batch + (a single `_persist_events_and_state_updates` call), with the message + first, so that the message's stream ordering is the batch minimum.""" + assert self.persistence is not None + prev_event_ids = self.get_success( + self.store.get_prev_events_for_room(self.room_id) + ) + + message, message_ctx = self.get_success( + create_event( + self.hs, + room_id=self.room_id, + type="m.room.message", + sender=self.alice, + content={"msgtype": "m.text", "body": "batched message"}, + prev_event_ids=prev_event_ids, + ) + ) + state_event, state_ctx = self.get_success( + create_event( + self.hs, + room_id=self.room_id, + type="m.call.member", + state_key=self.alice, + sender=self.alice, + content={"memberships": [{"device_id": "BATCHED"}]}, + prev_event_ids=prev_event_ids, + ) + ) + + self.get_success( + self.persistence.persist_events( + [(message, message_ctx), (state_event, state_ctx)] + ) + ) + return message, state_event + + def _assert_state_after_is_current_state(self, room: JsonDict) -> None: + """For a joined room `end_token` is the global `now_token`, so every + entry in `state_after` must be the room's current state for that key. + + This is the guard against "fix" shapes that seed `state_after` straight + from the timeline: a state event can be in the timeline and *not* be the + state at the end of it. + """ + current_state = self.get_success( + self.store.get_partial_current_state_ids(self.room_id) + ) + current_ids = set(current_state.values()) + for event in room["org.matrix.msc4222.state_after"]["events"]: + self.assertIn( + event["event_id"], + current_ids, + f"state_after reports {event['type']}/{event['state_key']} = " + f"{event['event_id']}, which is not the current state " + f"({current_state.get((event['type'], event['state_key']))})", + ) + + # ------------------------------------------------------------------ + # The persist-batch window + # ------------------------------------------------------------------ + + def test_state_delta_stream_id_of_batched_state_event(self) -> None: + """Documents that `current_state_delta_stream.stream_id` for a state + event persisted in a batch is the *minimum* stream ordering of the batch + (see `_persist_events_txn`: `min_stream_order`), not the state event's + own stream ordering. + + This is the storage-level precondition for the "timeline has it, + state_after doesn't" symptom: any sync token that falls strictly between + the batch minimum and the state event's stream ordering will put the + state event in the timeline while excluding its delta. + """ + message, state_event = self._persist_batch() + + message_pos = message.internal_metadata.stream_ordering + state_pos = state_event.internal_metadata.stream_ordering + assert message_pos is not None and state_pos is not None + self.assertLess(message_pos, state_pos) + + rows = self.get_success( + self.store.db_pool.simple_select_list( + table="current_state_delta_stream", + keyvalues={"room_id": self.room_id, "type": "m.call.member"}, + retcols=("stream_id", "event_id", "instance_name"), + desc="test_state_delta_stream_id", + ) + ) + self.assertEqual(len(rows), 1, rows) + delta_stream_id, delta_event_id, _instance_name = rows[0] + self.assertEqual(delta_event_id, state_event.event_id) + + self.assertEqual( + delta_stream_id, + message_pos, + "expected the delta to be recorded at the batch minimum", + ) + + deltas = self.get_success( + self.store.get_current_state_deltas_for_room( + self.room_id, + from_token=RoomStreamToken(stream=message_pos), + to_token=RoomStreamToken(stream=state_pos), + ) + ) + self.assertEqual( + [d.event_id for d in deltas], + [], + "state delta unexpectedly visible in (message_pos, state_pos]", + ) + + def test_state_after_with_token_inside_persist_batch(self) -> None: + """The end-to-end consequence of the above: if a client's `since` token + lands strictly inside a persist batch (which a *reader* worker can + observe, because it advances its events-stream position from replication + RDATA batches that may split a persist batch), the state event is in the + timeline of the next sync and must also be in `state_after`. + """ + sync_url = self._sync_url(lazy_load_members=False) + base = self._sync(sync_url)["next_batch"] + base_token = self.get_success(StreamToken.from_string(self.store, base)) + + message, state_event = self._persist_batch() + message_pos = message.internal_metadata.stream_ordering + assert message_pos is not None + + # A token positioned just after the message but before the state event + # -- i.e. in the middle of the persist batch. + split_token = base_token.copy_and_replace( + StreamKeyType.ROOM, RoomStreamToken(stream=message_pos) + ) + + deltas_up_to_split = self.get_success( + self.store.get_current_state_deltas_for_room( + self.room_id, + from_token=base_token.room_key, + to_token=RoomStreamToken(stream=message_pos), + ) + ) + self.assertIn( + state_event.event_id, + [d.event_id for d in deltas_up_to_split], + "expected the delta to be visible *before* its event", + ) + + split_since = self.get_success(split_token.to_string(self.store)) + room = self._joined_room(self._sync(sync_url, split_since)) + self.assertIn(state_event.event_id, self._timeline_ids(room)) + self.assertIn( + state_event.event_id, + self._state_after_ids(room), + f"state event in timeline but missing from state_after when the " + f"since token splits a persist batch: {room}", + ) + + def test_events_replication_stream_splits_persist_batches(self) -> None: + """Shows that the `since` token used by + `test_state_after_with_token_inside_persist_batch` is one a real + deployment hands out. + + The events replication stream emits a distinct token for every event + stream ordering (`_batch_updates` only collapses rows that share a + token), and `_process_rdata` calls `on_rdata` -- and hence + `process_replication_position` -- once per token. So a reader worker + (e.g. a sync worker) advances its events-stream position *through* the + middle of a persist batch, one event at a time, while reading + `current_state_delta_stream` straight from the database. + """ + before = self.store.get_room_max_token().stream + message, state_event = self._persist_batch() + after = self.store.get_room_max_token().stream + + message_pos = message.internal_metadata.stream_ordering + state_pos = state_event.internal_metadata.stream_ordering + assert message_pos is not None and state_pos is not None + + stream = EventsStream(self.hs) + updates, _upto, _limited = self.get_success( + # A limit of 100 is comfortably more than the handful of rows this + # test persists, so the update batch is never truncated. + stream._update_function("master", before, after, 100) + ) + + # The tokens a reader will actually advance to, in order. + delivered = [token for token, _row in _batch_updates(updates) if token] + + self.assertIn( + message_pos, + delivered, + "a reader worker advances to the batch minimum before it sees the " + "state event", + ) + self.assertIn(state_pos, delivered) + self.assertLess(delivered.index(message_pos), delivered.index(state_pos)) + + # ------------------------------------------------------------------ + # Guards: `state_after` must be the *resolved* state, not a replay of + # the timeline + # ------------------------------------------------------------------ + + def test_state_res_loser_alone_in_timeline(self) -> None: + """A state event that appears in the timeline but *loses* state + resolution must not be reported in `state_after` -- `state_after` is + the resolved state at the end of the timeline, not a replay of the + timeline. + + Here the client has already synced past the first of two conflicting + events, so the second one arrives alone in the timeline with no delta of + its own. + """ + sync_url = self._sync_url(lazy_load_members=False) + + fork_point = self.get_success(self.store.get_prev_events_for_room(self.room_id)) + + # Pin the state resolution outcome: conflicted events at the same + # mainline position are applied in `(origin_server_ts, event_id)` + # order with the last application winning, so giving the first event + # the later timestamp makes it deterministically beat the second one. + now = self.clock.time_msec() + + first = self.get_success( + inject_event( + self.hs, + room_id=self.room_id, + type="m.call.member", + state_key=self.alice, + sender=self.alice, + content={"memberships": [{"device_id": "FIRST"}]}, + prev_event_ids=fork_point, + origin_server_ts=now + 1000, + ) + ) + + # The client syncs past the first event. + since = self._sync(sync_url)["next_batch"] + + # A conflicting event forked off the same point arrives afterwards. + second = self.get_success( + inject_event( + self.hs, + room_id=self.room_id, + type="m.call.member", + state_key=self.alice, + sender=self.alice, + content={"memberships": [{"device_id": "SECOND"}]}, + prev_event_ids=fork_point, + origin_server_ts=now, + ) + ) + + current_state = self.get_success( + self.store.get_partial_current_state_ids(self.room_id) + ) + self.assertEqual( + current_state[("m.call.member", self.alice)], + first.event_id, + "test setup: expected the first event to win state resolution", + ) + + room = self._joined_room(self._sync(sync_url, since)) + self.assertIn(second.event_id, self._timeline_ids(room)) + self._assert_state_after_is_current_state(room) + self.assertNotIn( + second.event_id, + self._state_after_ids(room), + "a state-res loser leaked into state_after", + ) + + def test_state_after_is_current_state_with_split_token(self) -> None: + """Whatever mechanism repairs the persist-batch window must not report + anything other than the state at the end of the timeline.""" + sync_url = self._sync_url(lazy_load_members=False) + base = self._sync(sync_url)["next_batch"] + base_token = self.get_success(StreamToken.from_string(self.store, base)) + + message, state_event = self._persist_batch() + message_pos = message.internal_metadata.stream_ordering + assert message_pos is not None + + split_token = base_token.copy_and_replace( + StreamKeyType.ROOM, RoomStreamToken(stream=message_pos) + ) + since = self.get_success(split_token.to_string(self.store)) + + room = self._joined_room(self._sync(sync_url, since)) + self._assert_state_after_is_current_state(room) + self.assertIn(state_event.event_id, self._state_after_ids(room)) + + def test_gappy_timeline_with_split_token(self) -> None: + """The gappy variant of the persist-batch window: the `since` token + splits a persist batch *and* the timeline is truncated so that the + state event falls outside the window. The state event is then in + neither the timeline nor the deltas, but it is a state change since + `since`, so `state_after` must still report it -- with `limited: true` + the client relies entirely on `state_after` to bridge the gap. + """ + sync_url = self._sync_url(lazy_load_members=False, timeline_limit=3) + base = self._sync(sync_url)["next_batch"] + base_token = self.get_success(StreamToken.from_string(self.store, base)) + + message, state_event = self._persist_batch() + message_pos = message.internal_metadata.stream_ordering + assert message_pos is not None + + # Push the state event out of the timeline window. + for i in range(10): + self.helper.send(self.room_id, body=f"filler {i}", tok=self.bob_tok) + + split_token = base_token.copy_and_replace( + StreamKeyType.ROOM, RoomStreamToken(stream=message_pos) + ) + since = self.get_success(split_token.to_string(self.store)) + + room = self._joined_room(self._sync(sync_url, since)) + self.assertTrue(room["timeline"].get("limited"), room) + self.assertNotIn(state_event.event_id, self._timeline_ids(room)) + self.assertIn( + state_event.event_id, + self._state_after_ids(room), + f"state event outside a truncated timeline is missing from " + f"state_after when the since token splits a persist batch: {room}", + ) From 743b2feb7533ff66f67a617efabbfb383d947887 Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Mon, 31 Aug 2026 12:09:03 +0200 Subject: [PATCH 2/5] MSC4222: bound state deltas on the state event's own stream position current_state_delta_stream rows are stamped with the *minimum* stream ordering of the persist batch of their event, so bounding the MSC4222 delta query on the row's stream_id drops the deltas of a batch's state events for any since token that falls inside the batch. A worker reading the events stream from replication routinely observes such a token, since RDATA advances the stream one event at a time; the state event is then in the sync timeline but missing from state_after, or -- when the timeline is truncated -- silently missing altogether. Add get_current_state_deltas_for_room_by_event_position, which bounds each delta on the maximum of the row's stream_id and its event's own stream ordering: - a state event persisted mid-batch is tracked at its own position - rows with no event (the last-local-user state clearance) keep the row's position - rows stamped after their event (the partial-state resync path, which re-announces existing state at a fresh position) also keep the row's position, preserving the re-announcement That maximum is not a bound an index can serve, so the window is fetched as the union of two index-driven sets, with exact per-writer filtering in Python via _filter_results_by_stream as the existing query does: rows whose own stream_id is in the window (an index range on (room_id, stream_id), the same cost as the existing query), and rows whose *event* is in the window (driven by the events (room_id, stream_ordering) index over the window's state events, joined back via a new partial index on current_state_delta_stream(event_id) -- a row stamped below the window with an effective position inside it must have its event inside the window). Overall cost is proportional to the window, as before. The early-return optimisation consults the events stream cache as well as the delta stream cache, since a delta's effective position can now exceed the row stamp the delta cache tracks. Only the MSC4222 sync path uses the new method; every other consumer of current_state_delta_stream is untouched. --- changelog.d/pr-id.bugfix | 1 + synapse/handlers/sync.py | 14 +- .../storage/databases/main/state_deltas.py | 222 +++++++++++++++++- synapse/storage/databases/main/stream.py | 2 +- ...urrent_state_delta_stream_event_id_idx.sql | 23 ++ 5 files changed, 254 insertions(+), 8 deletions(-) create mode 100644 changelog.d/pr-id.bugfix create mode 100644 synapse/storage/schema/main/delta/94/09_current_state_delta_stream_event_id_idx.sql diff --git a/changelog.d/pr-id.bugfix b/changelog.d/pr-id.bugfix new file mode 100644 index 00000000000..ab54bf8df43 --- /dev/null +++ b/changelog.d/pr-id.bugfix @@ -0,0 +1 @@ +Fix state events being omitted from the [MSC4222](https://github.com/matrix-org/matrix-spec-proposals/pull/4222) `state_after` sync response when the client's `since` token falls inside an event persistence batch, as could happen on worker deployments. diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 943105415a4..3649b74d881 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -1397,8 +1397,8 @@ async def _compute_state_delta_for_full_sync( # Now roll back the state by looking at the state deltas between # end_token and now. - deltas = await self.store.get_current_state_deltas_for_room( - room_id, + deltas = await self.store.get_current_state_deltas_for_room_by_event_position( + room_id=room_id, from_token=end_token.room_key, to_token=self.store.get_room_max_token(), ) @@ -1536,10 +1536,12 @@ async def _compute_state_delta_for_incremental_sync( # # i.e. we return all state deltas, including membership changes that # we'd normally exclude due to LL. - deltas = await self.store.get_current_state_deltas_for_room( - room_id=room_id, - from_token=since_token.room_key, - to_token=end_token.room_key, + deltas = ( + await self.store.get_current_state_deltas_for_room_by_event_position( + room_id=room_id, + from_token=since_token.room_key, + to_token=end_token.room_key, + ) ) for delta in deltas: if delta.event_id is None: diff --git a/synapse/storage/databases/main/state_deltas.py b/synapse/storage/databases/main/state_deltas.py index a5d5407327c..16c545f03c6 100644 --- a/synapse/storage/databases/main/state_deltas.py +++ b/synapse/storage/databases/main/state_deltas.py @@ -59,8 +59,9 @@ class StateDelta: class StateDeltasStore(SQLBaseStore): # This class must be mixed in with a child class which provides the following - # attribute. TODO: can we get static analysis to enforce this? + # attributes. TODO: can we get static analysis to enforce this? _curr_state_delta_stream_cache: StreamChangeCache + _events_stream_cache: StreamChangeCache def __init__( self, @@ -76,6 +77,13 @@ def __init__( table="current_state_delta_stream", columns=("room_id", "stream_id"), ) + self.db_pool.updates.register_background_index_update( + update_name="current_state_delta_stream_event_id_index", + index_name="current_state_delta_stream_event_id_idx", + table="current_state_delta_stream", + columns=("event_id",), + where_clause="event_id IS NOT NULL", + ) async def get_partial_current_state_deltas( self, prev_stream_id: int, max_stream_id: int, limit: int = 100 @@ -300,6 +308,218 @@ async def get_current_state_deltas_for_room( to_token=to_token, ) + def get_current_state_deltas_for_room_by_event_position_txn( + self, + txn: LoggingTransaction, + room_id: str, + *, + from_token: RoomStreamToken | None, + to_token: RoomStreamToken | None, + ) -> list[StateDelta]: + """ + Get the state deltas between two tokens, bounding each delta on the + position of its state event in the events stream rather than on the + delta row's own `stream_id`. + + (> `from_token` and <= `to_token`; results are ordered by that + effective position.) + + `current_state_delta_stream` rows are stamped with the *minimum* + stream ordering of the persist batch of their event (see + `_update_current_state_txn`), so bounding on `stream_id` alone drops + the deltas of a batch's state events for any token that falls inside + the batch -- a position that a worker reading the events stream from + replication routinely observes, since RDATA advances the stream one + event at a time. + + A delta's effective position is therefore taken to be the *maximum* of + the row's `stream_id` and its event's own stream ordering: + + * for a state event persisted mid-batch, that is the event's own + position, so the delta tracks the event exactly; + * for rows with no event (e.g. the state clearance when the last + local user leaves), the row's `stream_id` stands; + * for rows stamped *after* their event (the partial-state room resync + in `update_current_state` re-announces existing state at a fresh + position), the row's `stream_id` stands, preserving the + re-announcement. + + That maximum is not a bound an index can serve, so the window is + fetched as the union of two index-driven sets (with the exact + per-writer filtering done in Python, as for + `get_current_state_deltas_for_room_txn`): + + * rows whose own `stream_id` is in the window -- an index range on + `current_state_delta_stream(room_id, stream_id)`, exactly like + `get_current_state_deltas_for_room_txn`; this is every row except + the mid-batch stragglers; + * rows whose *event* is in the window -- driven by the + `events(room_id, stream_ordering)` index over the events in the + window, joined back via the partial + `current_state_delta_stream(event_id)` index (only state events + find a delta row; we deliberately don't pre-filter on + `e.state_key`, which is only backfilled from schema version 76). A + row stamped below the window with an effective position inside it + must have its event inside the window, so this query is what recovers + the mid-batch stragglers, at a cost proportional to the number of + events in the window. + """ + args: list[str | int] = [room_id] + + stream_id_from_clause = "" + if from_token is not None: + stream_id_from_clause = "AND ? < d.stream_id" + args.append(from_token.stream) + + stream_id_to_clause = "" + if to_token is not None: + stream_id_to_clause = "AND d.stream_id <= ?" + args.append(to_token.get_max_stream_pos()) + + # Rows below the window's lower bound can only have an effective + # position inside the window via their event, so the event-driven query is + # only needed when there is a lower bound at all. + by_event_position_sql = "" + if from_token is not None: + event_position_to_clause = "" + if to_token is not None: + event_position_to_clause = "AND e.stream_ordering <= ?" + + by_event_position_sql = f""" + UNION + SELECT d.instance_name, d.stream_id, d.type, d.state_key, + d.event_id, d.prev_event_id, + e.instance_name, e.stream_ordering + FROM events AS e + INNER JOIN current_state_delta_stream AS d + ON d.event_id = e.event_id AND d.room_id = e.room_id + WHERE e.room_id = ? + AND ? < e.stream_ordering {event_position_to_clause} + """ + args.extend([room_id, from_token.stream]) + if to_token is not None: + args.append(to_token.get_max_stream_pos()) + + sql = f""" + SELECT d.instance_name, d.stream_id, d.type, d.state_key, + d.event_id, d.prev_event_id, + e.instance_name, e.stream_ordering + FROM current_state_delta_stream AS d + LEFT JOIN events AS e ON e.event_id = d.event_id + WHERE d.room_id = ? {stream_id_from_clause} {stream_id_to_clause} + {by_event_position_sql} + """ + txn.execute(sql, args) + + deltas = [] + for row in txn: + ( + row_instance, + row_stream, + event_type, + state_key, + event_id, + prev_event_id, + event_instance, + event_stream, + ) = row + + # The effective position: the row's own stamp, unless the event + # sits later in the stream. + if event_stream is not None and event_stream > row_stream: + effective_instance, effective_stream = event_instance, event_stream + else: + effective_instance, effective_stream = row_instance, row_stream + + if _filter_results_by_stream( + from_token, to_token, effective_instance, effective_stream + ): + deltas.append( + ( + effective_stream, + StateDelta( + stream_id=row_stream, + room_id=room_id, + event_type=event_type, + state_key=state_key, + event_id=event_id, + prev_event_id=prev_event_id, + ), + ) + ) + + # Consumers rely on deltas being in stream order (the last delta for a + # given state key wins), which for this query means effective-position + # order. + deltas.sort(key=lambda t: t[0]) + return [d for _, d in deltas] + + @trace + async def get_current_state_deltas_for_room_by_event_position( + self, + room_id: str, + *, + from_token: RoomStreamToken | None, + to_token: RoomStreamToken | None, + ) -> list[StateDelta]: + """ + Get the state deltas between two tokens, bounding each delta on the + position of its state event rather than on the delta row's `stream_id`. + See `get_current_state_deltas_for_room_by_event_position_txn`. + + (> `from_token` and <= `to_token`) + + Until the `current_state_delta_stream(event_id)` index has been built + (a background update), this falls back to bounding on the rows' own + `stream_id` -- the behaviour this method replaces, which can miss + mid-batch deltas but never scans beyond the index. + """ + # We can bail early if the `from_token` is after the `to_token` + if ( + to_token is not None + and from_token is not None + and to_token.is_before_or_eq(from_token) + ): + return [] + + # A delta's effective position is beyond `from_token` only if the row's + # `stream_id` is (the delta stream cache) or its event's stream + # ordering is (the events stream cache); if neither cache has seen the + # room change there is nothing to return. + if ( + from_token is not None + and not self._curr_state_delta_stream_cache.has_entity_changed( + room_id, from_token.stream + ) + and not self._events_stream_cache.has_entity_changed( + room_id, from_token.stream + ) + ): + return [] + + # Without the `current_state_delta_stream(event_id)` index, the + # event-driven query of the union has no index to join through and would + # walk the room's entire delta history, so fall back to the plain + # `stream_id` bounds until the background update has completed. + if not await self.db_pool.updates.has_completed_background_update( + "current_state_delta_stream_event_id_index" + ): + return await self.db_pool.runInteraction( + "get_current_state_deltas_for_room_by_event_position_fallback", + self.get_current_state_deltas_for_room_txn, + room_id, + from_token=from_token, + to_token=to_token, + ) + + return await self.db_pool.runInteraction( + "get_current_state_deltas_for_room_by_event_position", + self.get_current_state_deltas_for_room_by_event_position_txn, + room_id, + from_token=from_token, + to_token=to_token, + ) + @trace async def get_current_state_deltas_for_rooms( self, diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 7d14f9f4d80..1d086010717 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -608,7 +608,7 @@ def __init__( stream_column="stream_ordering", max_value=events_max, ) - self._events_stream_cache = StreamChangeCache( + self._events_stream_cache: StreamChangeCache = StreamChangeCache( name="EventsRoomStreamChangeCache", server_name=self.server_name, current_stream_pos=min_event_val, diff --git a/synapse/storage/schema/main/delta/94/09_current_state_delta_stream_event_id_idx.sql b/synapse/storage/schema/main/delta/94/09_current_state_delta_stream_event_id_idx.sql new file mode 100644 index 00000000000..87c7bb851d7 --- /dev/null +++ b/synapse/storage/schema/main/delta/94/09_current_state_delta_stream_event_id_idx.sql @@ -0,0 +1,23 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2026 Element Creations, Ltd +-- +-- This program is free software: you can redistribute it and/or modify +-- it under the terms of the GNU Affero General Public License as +-- published by the Free Software Foundation, either version 3 of the +-- License, or (at your option) any later version. +-- +-- See the GNU Affero General Public License for more details: +-- . + + +-- Add an index on `current_state_delta_stream(event_id)` so that the deltas +-- of the state events in a sync window can be looked up by event, even when +-- the rows are stamped before the window (rows are stamped with the minimum +-- stream ordering of their persist batch, see `_update_current_state_txn`). +-- +-- This is a partial index as rows with a NULL event_id (state deletions) are +-- never looked up by event. +INSERT INTO background_updates (ordering, update_name, progress_json) VALUES + (9409, 'current_state_delta_stream_event_id_index', '{}'); From b296c32043ed4d836ea3227ba8dcd455bd6ebbac Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Mon, 31 Aug 2026 15:07:13 +0200 Subject: [PATCH 3/5] MSC4222: restrict the event-driven delta query to state events --- .../storage/databases/main/state_deltas.py | 34 ++++++++++++++----- 1 file changed, 26 insertions(+), 8 deletions(-) diff --git a/synapse/storage/databases/main/state_deltas.py b/synapse/storage/databases/main/state_deltas.py index 16c545f03c6..03e6eda1f40 100644 --- a/synapse/storage/databases/main/state_deltas.py +++ b/synapse/storage/databases/main/state_deltas.py @@ -315,6 +315,7 @@ def get_current_state_deltas_for_room_by_event_position_txn( *, from_token: RoomStreamToken | None, to_token: RoomStreamToken | None, + events_state_key_populated: bool = True, ) -> list[StateDelta]: """ Get the state deltas between two tokens, bounding each delta on the @@ -356,13 +357,11 @@ def get_current_state_deltas_for_room_by_event_position_txn( * rows whose *event* is in the window -- driven by the `events(room_id, stream_ordering)` index over the events in the window, joined back via the partial - `current_state_delta_stream(event_id)` index (only state events - find a delta row; we deliberately don't pre-filter on - `e.state_key`, which is only backfilled from schema version 76). A - row stamped below the window with an effective position inside it - must have its event inside the window, so this query is what recovers - the mid-batch stragglers, at a cost proportional to the number of - events in the window. + `current_state_delta_stream(event_id)` index. A row stamped below + the window with an effective position inside it must have its event + inside the window, so this query is what recovers the mid-batch + stragglers, at a cost proportional to the number of state events in + the window. """ args: list[str | int] = [room_id] @@ -385,6 +384,13 @@ def get_current_state_deltas_for_room_by_event_position_txn( if to_token is not None: event_position_to_clause = "AND e.stream_ordering <= ?" + # Only state events can match the delta join, so once + # `events.state_key` is reliable we restrict the scan to them and + # spare one index probe per non-state event in the window. + event_state_key_clause = "" + if events_state_key_populated: + event_state_key_clause = "AND e.state_key IS NOT NULL" + by_event_position_sql = f""" UNION SELECT d.instance_name, d.stream_id, d.type, d.state_key, @@ -393,7 +399,7 @@ def get_current_state_deltas_for_room_by_event_position_txn( FROM events AS e INNER JOIN current_state_delta_stream AS d ON d.event_id = e.event_id AND d.room_id = e.room_id - WHERE e.room_id = ? + WHERE e.room_id = ? {event_state_key_clause} AND ? < e.stream_ordering {event_position_to_clause} """ args.extend([room_id, from_token.stream]) @@ -512,12 +518,24 @@ async def get_current_state_deltas_for_room_by_event_position( to_token=to_token, ) + # `events.state_key` is back-populated by a schema-76 background + # update; until it has completed, old state events may have a NULL + # state_key and the event-driven query must not filter on it. + # (`has_completed_background_update` memoises completion, so this is + # only a query the first time.) + events_state_key_populated = ( + await self.db_pool.updates.has_completed_background_update( + "events_populate_state_key_rejections" + ) + ) + return await self.db_pool.runInteraction( "get_current_state_deltas_for_room_by_event_position", self.get_current_state_deltas_for_room_by_event_position_txn, room_id, from_token=from_token, to_token=to_token, + events_state_key_populated=events_state_key_populated, ) @trace From 509321337f228640397ed635810e4241614e2e71 Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Mon, 31 Aug 2026 15:09:04 +0200 Subject: [PATCH 4/5] MSC4222: add storage-level tests for the by-event-position delta query --- tests/storage/test_state_deltas.py | 218 +++++++++++++++++++++++++++++ 1 file changed, 218 insertions(+) create mode 100644 tests/storage/test_state_deltas.py diff --git a/tests/storage/test_state_deltas.py b/tests/storage/test_state_deltas.py new file mode 100644 index 00000000000..08eb1d7090a --- /dev/null +++ b/tests/storage/test_state_deltas.py @@ -0,0 +1,218 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations, Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# + +from twisted.test.proto_helpers import MemoryReactor + +import synapse.rest.admin +from synapse.events import EventBase +from synapse.rest.client import login, room +from synapse.server import HomeServer +from synapse.types import RoomStreamToken +from synapse.util.clock import Clock + +from tests import unittest +from tests.test_utils.event_injection import create_event + + +class StateDeltasByEventPositionTestCase(unittest.HomeserverTestCase): + """Tests for `get_current_state_deltas_for_room_by_event_position`, which + bounds each delta on the position of its state event rather than on the + delta row's own `stream_id` (rows are stamped with the minimum stream + ordering of their persist batch, so the two can differ).""" + + servlets = [ + synapse.rest.admin.register_servlets, + room.register_servlets, + login.register_servlets, + ] + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + self.persistence = hs.get_storage_controllers().persistence + assert self.persistence is not None + + self.alice = self.register_user("alice", "password") + self.alice_tok = self.login("alice", "password") + self.room_id = self.helper.create_room_as(self.alice, tok=self.alice_tok) + + def _persist_batch(self) -> tuple[EventBase, EventBase]: + """Persist a message and a state event in a single persist batch, with + the message first, so that the message's stream ordering is the batch + minimum and the state event's delta row is stamped before the state + event itself.""" + assert self.persistence is not None + prev_event_ids = self.get_success( + self.store.get_prev_events_for_room(self.room_id) + ) + + message, message_ctx = self.get_success( + create_event( + self.hs, + room_id=self.room_id, + type="m.room.message", + sender=self.alice, + content={"msgtype": "m.text", "body": "batched message"}, + prev_event_ids=prev_event_ids, + ) + ) + state_event, state_ctx = self.get_success( + create_event( + self.hs, + room_id=self.room_id, + type="m.call.member", + state_key=self.alice, + sender=self.alice, + content={"memberships": [{"device_id": "BATCHED"}]}, + prev_event_ids=prev_event_ids, + ) + ) + + self.get_success( + self.persistence.persist_events( + [(message, message_ctx), (state_event, state_ctx)] + ) + ) + return message, state_event + + def _batch_positions(self) -> tuple[str, int, int]: + """Persist a batch and return (state event id, batch minimum position, + state event position), sanity-checking the batch shape.""" + message, state_event = self._persist_batch() + message_pos = message.internal_metadata.stream_ordering + state_pos = state_event.internal_metadata.stream_ordering + assert message_pos is not None and state_pos is not None + self.assertLess(message_pos, state_pos) + return state_event.event_id, message_pos, state_pos + + def test_mid_batch_delta_is_in_window(self) -> None: + """A window whose lower bound splits a persist batch contains the + batch's state event delta, which the stamp-bounded query misses.""" + state_event_id, message_pos, state_pos = self._batch_positions() + + from_token = RoomStreamToken(stream=message_pos) + to_token = RoomStreamToken(stream=state_pos) + + # The stamp-bounded query misses the delta: the row is stamped at the + # batch minimum, below the window. + deltas = self.get_success( + self.store.get_current_state_deltas_for_room( + self.room_id, from_token=from_token, to_token=to_token + ) + ) + self.assertEqual([d.event_id for d in deltas], []) + + # The by-event-position query recovers it. + deltas = self.get_success( + self.store.get_current_state_deltas_for_room_by_event_position( + self.room_id, from_token=from_token, to_token=to_token + ) + ) + self.assertEqual([d.event_id for d in deltas], [state_event_id]) + + def test_delta_is_not_reported_before_its_event(self) -> None: + """A window ending at the batch minimum must not contain the state + event's delta: its effective position is the event's own, beyond the + window. (The stamp-bounded query reports it here, one window early.)""" + state_event_id, message_pos, _state_pos = self._batch_positions() + + to_token = RoomStreamToken(stream=message_pos) + + deltas = self.get_success( + self.store.get_current_state_deltas_for_room( + self.room_id, from_token=None, to_token=to_token + ) + ) + self.assertIn(state_event_id, [d.event_id for d in deltas]) + + deltas = self.get_success( + self.store.get_current_state_deltas_for_room_by_event_position( + self.room_id, from_token=None, to_token=to_token + ) + ) + self.assertNotIn(state_event_id, [d.event_id for d in deltas]) + + def test_no_lower_bound(self) -> None: + """With no lower bound the event-driven query is unnecessary and the + query returns everything up to the upper bound, batch rows included.""" + state_event_id, _message_pos, state_pos = self._batch_positions() + + deltas = self.get_success( + self.store.get_current_state_deltas_for_room_by_event_position( + self.room_id, + from_token=None, + to_token=RoomStreamToken(stream=state_pos), + ) + ) + # The room's creation state plus the batched state event. + self.assertIn(state_event_id, [d.event_id for d in deltas]) + + def test_no_upper_bound(self) -> None: + """With no upper bound a mid-batch lower bound still recovers the + batch's state event delta.""" + state_event_id, message_pos, _state_pos = self._batch_positions() + + deltas = self.get_success( + self.store.get_current_state_deltas_for_room_by_event_position( + self.room_id, + from_token=RoomStreamToken(stream=message_pos), + to_token=None, + ) + ) + self.assertEqual([d.event_id for d in deltas], [state_event_id]) + + def test_unfiltered_event_driven_query(self) -> None: + """Until `events.state_key` has been back-populated the event-driven + query cannot filter on it; the unfiltered mode must find the same + deltas.""" + state_event_id, message_pos, state_pos = self._batch_positions() + + deltas = self.get_success( + self.store.db_pool.runInteraction( + "test_unfiltered_event_driven_query", + self.store.get_current_state_deltas_for_room_by_event_position_txn, + self.room_id, + from_token=RoomStreamToken(stream=message_pos), + to_token=RoomStreamToken(stream=state_pos), + events_state_key_populated=False, + ) + ) + self.assertEqual([d.event_id for d in deltas], [state_event_id]) + + def test_falls_back_until_index_built(self) -> None: + """Until the `current_state_delta_stream(event_id)` index has been + built, the query falls back to the stamp-bounded behaviour (which can + miss mid-batch deltas but never scans without an index).""" + state_event_id, message_pos, state_pos = self._batch_positions() + + # Pretend the index's background update is still pending. + self.get_success( + self.store.db_pool.simple_insert( + "background_updates", + { + "update_name": "current_state_delta_stream_event_id_index", + "progress_json": "{}", + }, + ) + ) + self.store.db_pool.updates._all_done = False + + deltas = self.get_success( + self.store.get_current_state_deltas_for_room_by_event_position( + self.room_id, + from_token=RoomStreamToken(stream=message_pos), + to_token=RoomStreamToken(stream=state_pos), + ) + ) + # Stamp-bounded behaviour: the mid-batch delta is missed. + self.assertEqual([d.event_id for d in deltas], []) From e4481ecea57d1f42bc7d0fec50971ca6c071099b Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Mon, 31 Aug 2026 17:56:19 +0200 Subject: [PATCH 5/5] Rename changelog fragment to the PR number --- changelog.d/{pr-id.bugfix => 20171.bugfix} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename changelog.d/{pr-id.bugfix => 20171.bugfix} (100%) diff --git a/changelog.d/pr-id.bugfix b/changelog.d/20171.bugfix similarity index 100% rename from changelog.d/pr-id.bugfix rename to changelog.d/20171.bugfix