Fix 404 on `/sync` when the last event is a redaction of an unknown/purged event (#12905)

Currently, we try to pull the event corresponding to a sync token from the database. However, when
we fetch redaction events, we check the target of that redaction (because we aren't allowed to send
redactions to clients without validating them). So, if the sync token points to a redaction of an event
that we don't have, we have a problem.

It turns out we don't really need that event, and can just work with its ID and metadata, which
sidesteps the whole problem.

2 files changed, 14 insertions, 10 deletions

diff --git a/synapse/storage/databases/main/state.py b/synapse/storage/databases/main/state.py
index a07ad85582..3f2be3854b 100644
--- a/synapse/storage/databases/main/state.py
+++ b/synapse/storage/databases/main/state.py
@@ -54,6 +54,7 @@ class EventMetadata:
     room_id: str
     event_type: str
     state_key: Optional[str]
+    rejection_reason: Optional[str]
 
 
 def _retrieve_and_check_room_version(room_id: str, room_version_id: str) -> RoomVersion:
@@ -167,17 +168,22 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
             )
 
             sql = f"""
-                SELECT e.event_id, e.room_id, e.type, se.state_key FROM events AS e
+                SELECT e.event_id, e.room_id, e.type, se.state_key, r.reason
+                FROM events AS e
                 LEFT JOIN state_events se USING (event_id)
+                LEFT JOIN rejections r USING (event_id)
                 WHERE {clause}
             """
 
             txn.execute(sql, args)
             return {
                 event_id: EventMetadata(
-                    room_id=room_id, event_type=event_type, state_key=state_key
+                    room_id=room_id,
+                    event_type=event_type,
+                    state_key=state_key,
+                    rejection_reason=rejection_reason,
                 )
-                for event_id, room_id, event_type, state_key in txn
+                for event_id, room_id, event_type, state_key, rejection_reason in txn
             }
 
         result_map: Dict[str, EventMetadata] = {}
diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py
index 0e3a23a140..8e88784d3c 100644
--- a/synapse/storage/databases/main/stream.py
+++ b/synapse/storage/databases/main/stream.py
@@ -765,15 +765,16 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         self,
         room_id: str,
         end_token: RoomStreamToken,
-    ) -> Optional[EventBase]:
-        """Returns the last event in a room at or before a stream ordering
+    ) -> Optional[str]:
+        """Returns the ID of the last event in a room at or before a stream ordering
 
         Args:
             room_id
             end_token: The token used to stream from
 
         Returns:
-            The most recent event.
+            The ID of the most recent event, or None if there are no events in the room
+            before this stream ordering.
         """
 
         last_row = await self.get_room_event_before_stream_ordering(
@@ -781,10 +782,7 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             stream_ordering=end_token.stream,
         )
         if last_row:
-            _, _, event_id = last_row
-            event = await self.get_event(event_id, get_prev_content=True)
-            return event
-
+            return last_row[2]
         return None
 
     async def get_current_room_stream_token_for_room_id(