Be smarter about which hosts to send presence to when processing room joins (#9402)

This PR attempts to eliminate unnecessary presence sending work when your local server joins a room, or when a remote server joins a room your server is participating in by processing state deltas in chunks rather than individually.

---

When your server joins a room for the first time, it requests the historical state as well. This chunk of new state is passed to the presence handler which, after filtering that state down to only membership joins, will send presence updates to homeservers for each join processed.

It turns out that we were being a bit naive and processing each event individually, and sending out presence updates for every one of those joins. Even if many different joins were users on the same server (hello IRC bridges), we'd send presence to that same homeserver for every remote user join we saw.

This PR attempts to deduplicate all of that by processing the entire batch of state deltas at once, instead of only doing each join individually. We process the joins and note down which servers need which presence:

* If it was a local user join, send that user's latest presence to all servers in the room
* If it was a remote user join, send the presence for all local users in the room to that homeserver

We deduplicate by inserting all of those pending updates into a dictionary of the form:

```
{
  server_name1: {presence_update1, ...},
  server_name2: {presence_update1, presence_update2, ...}
}
```

Only after building this dict do we then start sending out presence updates.

1 files changed, 10 insertions, 4 deletions

diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py
index be2ee26f07..996c614198 100644
--- a/tests/handlers/test_presence.py
+++ b/tests/handlers/test_presence.py
@@ -521,7 +521,7 @@ class PresenceJoinTestCase(unittest.HomeserverTestCase):
         )
         self.assertEqual(expected_state.state, PresenceState.ONLINE)
         self.federation_sender.send_presence_to_destinations.assert_called_once_with(
-            destinations=["server2"], states=[expected_state]
+            destinations=["server2"], states={expected_state}
         )
 
         #
@@ -533,7 +533,7 @@ class PresenceJoinTestCase(unittest.HomeserverTestCase):
 
         self.federation_sender.send_presence.assert_not_called()
         self.federation_sender.send_presence_to_destinations.assert_called_once_with(
-            destinations=["server3"], states=[expected_state]
+            destinations=["server3"], states={expected_state}
         )
 
     def test_remote_gets_presence_when_local_user_joins(self):
@@ -584,8 +584,14 @@ class PresenceJoinTestCase(unittest.HomeserverTestCase):
             self.presence_handler.current_state_for_user("@test2:server")
         )
         self.assertEqual(expected_state.state, PresenceState.ONLINE)
-        self.federation_sender.send_presence_to_destinations.assert_called_once_with(
-            destinations={"server2", "server3"}, states=[expected_state]
+        self.assertEqual(
+            self.federation_sender.send_presence_to_destinations.call_count, 2
+        )
+        self.federation_sender.send_presence_to_destinations.assert_any_call(
+            destinations=["server3"], states={expected_state}
+        )
+        self.federation_sender.send_presence_to_destinations.assert_any_call(
+            destinations=["server2"], states={expected_state}
         )
 
     def _add_new_user(self, room_id, user_id):