Fix state events missing from MSC4222 state_after when the since token falls inside a persist batch

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

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:

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,236 @@ 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,
+        events_state_key_populated: bool = True,
+    ) -> 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. 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]
+
+        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 <= ?"
+
+            # 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,
+                    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 = ? {event_state_key_clause}
+                    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,
+            )
+
+        # `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
     async def get_current_state_deltas_for_rooms(
         self,

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,

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:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+
+-- 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', '{}');