Fix bug where we didn't correctly filter lazy members by room

When fetching previously sent lazy members we didn't filter by room,
which meant that we didn't send down member events in a room if we'd
previously sent that user's member event in another room.

Author: erikjohnston

synapse/storage/databases/main/sliding_sync.py

@@ -25,6 +25,7 @@
     DatabasePool,
     LoggingDatabaseConnection,
     LoggingTransaction,
+    make_in_list_sql_clause,
 )
 from synapse.storage.engines import PostgresEngine
 from synapse.types import MultiWriterStreamToken, RoomStreamToken
@@ -532,14 +533,17 @@ async def get_sliding_sync_connection_lazy_members(
         def get_sliding_sync_connection_lazy_members_txn(
             txn: LoggingTransaction,
         ) -> Mapping[str, int]:
-            sql = """
-                SELECT user_id, mem.connection_position, last_seen_ts
-                FROM sliding_sync_connection_positions AS pos
-                INNER JOIN sliding_sync_connection_lazy_members AS mem USING (connection_key)
-                WHERE pos.connection_position = ?
+            user_clause, user_args = make_in_list_sql_clause(
+                txn.database_engine, "user_id", user_ids
+            )
+
+            sql = f"""
+                SELECT user_id, connection_position, last_seen_ts
+                FROM sliding_sync_connection_lazy_members AS pos
+                WHERE room_id = ? AND {user_clause}
             """
 
-            txn.execute(sql, (connection_position,))
+            txn.execute(sql, (room_id, *user_args))
 
             # Filter out any cache entries that only apply to forked connection
             # positions. Entries with `NULL` connection position apply to all

tests/rest/client/sliding_sync/test_rooms_required_state.py

@@ -932,6 +932,92 @@ def test_lazy_members_limited_sync(self) -> None:
             exact=True,
         )
 
+    def test_lazy_members_across_multiple_rooms(self) -> None:
+        """Test that lazy loading room members are tracked per-room correctly.
+        """
+
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+        user2_id = self.register_user("user2", "pass")
+        user2_tok = self.login(user2_id, "pass")
+
+        # Create two rooms with both users in them and send a message in each
+        room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.join(room_id1, user1_id, tok=user1_tok)
+        self.helper.send(room_id1, "room1-msg1", tok=user2_tok)
+
+        room_id2 = self.helper.create_room_as(user2_id, tok=user2_tok)
+        self.helper.join(room_id2, user1_id, tok=user1_tok)
+        self.helper.send(room_id2, "room2-msg1", tok=user2_tok)
+
+        # Make a sync with lazy loading for the room members to establish
+        # a position
+        sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [
+                        [EventTypes.Member, StateValues.LAZY],
+                    ],
+                    "timeline_limit": 1,
+                }
+            }
+        }
+        response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # We expect to see only user2's membership in both rooms
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id1)
+        )
+        self._assertRequiredStateIncludes(
+            response_body["rooms"][room_id1]["required_state"],
+            {
+                state_map[(EventTypes.Member, user2_id)],
+            },
+            exact=True,
+        )
+
+        # Send a message in room1 from user1
+        self.helper.send(room_id1, "room1-msg2", tok=user1_tok)
+
+        # Make an incremental Sliding Sync request and check that we get user1's
+        # membership.
+        response_body, from_token = self.do_sync(
+            sync_body, since=from_token, tok=user1_tok
+        )
+
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id1)
+        )
+        self._assertRequiredStateIncludes(
+            response_body["rooms"][room_id1]["required_state"],
+            {
+                state_map[(EventTypes.Member, user1_id)],
+            },
+            exact=True,
+        )
+
+        # Send a message in room2 from user1
+        self.helper.send(room_id2, "room2-msg2", tok=user1_tok)
+
+        # Make an incremental Sliding Sync request and check that we get user1's
+        # membership.
+        response_body, from_token = self.do_sync(
+            sync_body, since=from_token, tok=user1_tok
+        )
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id2)
+        )
+        self._assertRequiredStateIncludes(
+            response_body["rooms"][room_id2]["required_state"],
+            {
+                state_map[(EventTypes.Member, user1_id)],
+            },
+            exact=True,
+        )
+
+
+
     def test_rooms_required_state_me(self) -> None:
         """
         Test `rooms.required_state` correctly handles $ME.