Actually test that event persistence is sharded

tests/replication/test_sharded_event_persister.py

@@ -18,6 +18,7 @@ from synapse.rest import admin
 from synapse.rest.client.v1 import login, room
 
 from tests.replication._base import BaseMultiWorkerStreamTestCase
+from tests.utils import USE_POSTGRES_FOR_TESTS
 
 logger = logging.getLogger(__name__)
 
@@ -26,6 +27,11 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
     """Checks event persisting sharding works
     """
 
+    # Event persister sharding requires postgres (due to needing
+    # `MutliWriterIdGenerator`).
+    if not USE_POSTGRES_FOR_TESTS:
+        skip = "Requires Postgres"
+
     servlets = [
         admin.register_servlets_for_client_rest_resource,
         room.register_servlets,
@@ -48,8 +54,8 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
         return conf
 
     def test_basic(self):
-        """Simple test that rooms can be created and joined when there are
-        multiple event persisters.
+        """Simple test to ensure that multiple rooms can be created and joined,
+        and that different rooms get handled by different instances.
         """
 
         self.make_worker_hs(
@@ -60,9 +66,17 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
             "synapse.app.generic_worker", {"worker_name": "worker2"},
         )
 
+        persisted_on_1 = False
+        persisted_on_2 = False
+
+        store = self.hs.get_datastore()
+
         user_id = self.register_user("user", "pass")
         access_token = self.login("user", "pass")
 
+        # Keep making new rooms until we see rooms being persisted on both
+        # workers.
+        for _ in range(10):
             # Create a room
             room = self.helper.create_room_as(user_id, tok=access_token)
 
@@ -72,4 +86,17 @@ class EventPersisterShardTestCase(BaseMultiWorkerStreamTestCase):
             )
 
             # The other user sends some messages
-        self.helper.send(room, body="Hi!", tok=self.other_access_token)
+            rseponse = self.helper.send(room, body="Hi!", tok=self.other_access_token)
+            event_id = rseponse["event_id"]
+
+            # The event position includes which instance persisted the event.
+            pos = self.get_success(store.get_position_for_event(event_id))
+
+            persisted_on_1 |= pos.instance_name == "worker1"
+            persisted_on_2 |= pos.instance_name == "worker2"
+
+            if persisted_on_1 and persisted_on_2:
+                break
+
+        self.assertTrue(persisted_on_1)
+        self.assertTrue(persisted_on_2)