Sliding sync: various fixups to the sliding sync joined room background job (#17673)

Follow-up to #17652, https://github.com/element-hq/synapse/pull/17641,
https://github.com/element-hq/synapse/pull/17634,
https://github.com/element-hq/synapse/pull/17631 and
https://github.com/element-hq/synapse/pull/17632 to fix-up
https://github.com/element-hq/synapse/pull/17512

changelog.d/17673.misc

@@ -0,0 +1 @@
+Pre-populate room data used in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint for quick filtering/sorting.

synapse/storage/databases/main/events_bg_updates.py

@@ -1595,17 +1595,15 @@ class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseS
             # starve disk usage while this goes on.
             #
             # We upsert in case we have to run this multiple times.
-            #
-            # The `WHERE TRUE` clause is to avoid "Parsing Ambiguity"
             txn.execute(
                 """
                 INSERT INTO sliding_sync_joined_rooms_to_recalculate
                     (room_id)
-                SELECT room_id FROM rooms WHERE ?
+                SELECT DISTINCT room_id FROM local_current_membership
+                WHERE membership = 'join'
                 ON CONFLICT (room_id)
                 DO NOTHING;
                 """,
-                (True,),
             )
 
         await self.db_pool.runInteraction(
@@ -1689,7 +1687,15 @@ class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseS
             if not current_state_ids_map:
                 continue
 
-            fetched_events = await self.get_events(current_state_ids_map.values())
+            try:
+                fetched_events = await self.get_events(current_state_ids_map.values())
+            except (DatabaseCorruptionError, InvalidEventError) as e:
+                logger.warning(
+                    "Failed to fetch state for room '%s' due to corrupted events. Ignoring. Error: %s",
+                    room_id,
+                    e,
+                )
+                continue
 
             current_state_map: StateMap[EventBase] = {
                 state_key: fetched_events[event_id]
@@ -1722,10 +1728,13 @@ class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseS
                 + "given we pulled the room out of `current_state_events`"
             )
             most_recent_event_stream_ordering = most_recent_event_pos_results[1].stream
-            assert most_recent_event_stream_ordering > 0, (
-                "We should have at-least one event in the room (our own join membership event for example) "
-                + "that isn't backfilled (negative `stream_ordering`)  if we are joined to the room."
-            )
+
+            # The `most_recent_event_stream_ordering` should be positive,
+            # however there are (very rare) rooms where that is not the case in
+            # the matrix.org database. It's not clear how they got into that
+            # state, but does mean that we cannot assert that the stream
+            # ordering is indeed positive.
+
             # Figure out the latest `bump_stamp` in the room. This could be `None` for a
             # federated room you just joined where all of events are still `outliers` or
             # backfilled history. In the Sliding Sync API, we default to the user's