Add `event.internal_metadata.instance_name` (#17300)

Add `event.internal_metadata.instance_name` (the worker instance that persisted the event) to go alongside the existing `event.internal_metadata.stream_ordering`.

`instance_name` is useful to properly compare and query for events with a token since you need to compare both the `stream_ordering` and `instance_name` against the vector clock/`instance_map` in the `RoomStreamToken`.

This is pre-requisite work and may be used in https://github.com/element-hq/synapse/pull/17293

Adding `event.internal_metadata.instance_name` was first mentioned in the initial Sliding Sync PR while pairing with @erikjohnston, see https://github.com/element-hq/synapse/pull/17187/commits/09609cb0dbca3a4cfd9fbf90cc962e765ec469c0#diff-5cd773fb307aa754bd3948871ba118b1ef0303f4d72d42a2d21e38242bf4e096R405-R410

3 files changed, 11 insertions, 3 deletions

diff --git a/tests/events/test_utils.py b/tests/events/test_utils.py

index d5ac66a6ed..30f8787758 100644
--- a/tests/events/test_utils.py
+++ b/tests/events/test_utils.py
@@ -625,6 +625,8 @@ class CloneEventTestCase(stdlib_unittest.TestCase):
         )
         original.internal_metadata.stream_ordering = 1234
         self.assertEqual(original.internal_metadata.stream_ordering, 1234)
+        original.internal_metadata.instance_name = "worker1"
+        self.assertEqual(original.internal_metadata.instance_name, "worker1")
 
         cloned = clone_event(original)
         cloned.unsigned["b"] = 3
@@ -632,6 +634,7 @@ class CloneEventTestCase(stdlib_unittest.TestCase):
         self.assertEqual(original.unsigned, {"a": 1, "b": 2})
         self.assertEqual(cloned.unsigned, {"a": 1, "b": 3})
         self.assertEqual(cloned.internal_metadata.stream_ordering, 1234)
+        self.assertEqual(cloned.internal_metadata.instance_name, "worker1")
         self.assertEqual(cloned.internal_metadata.txn_id, "txn")
 
 
diff --git a/tests/replication/storage/test_events.py b/tests/replication/storage/test_events.py

index 4e41a1c912..a56f1e2d5d 100644
--- a/tests/replication/storage/test_events.py
+++ b/tests/replication/storage/test_events.py
@@ -141,6 +141,7 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
         self.persist(type="m.room.create", key="", creator=USER_ID)
         self.check("get_invited_rooms_for_local_user", [USER_ID_2], [])
         event = self.persist(type="m.room.member", key=USER_ID_2, membership="invite")
+        assert event.internal_metadata.instance_name is not None
         assert event.internal_metadata.stream_ordering is not None
 
         self.replicate()
@@ -155,7 +156,7 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
                     "invite",
                     event.event_id,
                     PersistedEventPosition(
-                        self.hs.get_instance_name(),
+                        event.internal_metadata.instance_name,
                         event.internal_metadata.stream_ordering,
                     ),
                     RoomVersions.V1.identifier,
@@ -232,11 +233,12 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
         j2 = self.persist(
             type="m.room.member", sender=USER_ID_2, key=USER_ID_2, membership="join"
         )
+        assert j2.internal_metadata.instance_name is not None
         assert j2.internal_metadata.stream_ordering is not None
         self.replicate()
 
         expected_pos = PersistedEventPosition(
-            "master", j2.internal_metadata.stream_ordering
+            j2.internal_metadata.instance_name, j2.internal_metadata.stream_ordering
         )
         self.check(
             "get_rooms_for_user_with_stream_ordering",
@@ -288,6 +290,7 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
         msg, msgctx = self.build_event()
         self.get_success(self.persistance.persist_events([(j2, j2ctx), (msg, msgctx)]))
         self.replicate()
+        assert j2.internal_metadata.instance_name is not None
         assert j2.internal_metadata.stream_ordering is not None
 
         event_source = RoomEventSource(self.hs)
@@ -329,7 +332,8 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
             # joined_rooms list.
             if membership_changes:
                 expected_pos = PersistedEventPosition(
-                    "master", j2.internal_metadata.stream_ordering
+                    j2.internal_metadata.instance_name,
+                    j2.internal_metadata.stream_ordering,
                 )
                 self.assertEqual(
                     joined_rooms,
diff --git a/tests/storage/test_event_chain.py b/tests/storage/test_event_chain.py

index 27d5b0125f..81feb3ec29 100644
--- a/tests/storage/test_event_chain.py
+++ b/tests/storage/test_event_chain.py
@@ -431,6 +431,7 @@ class EventChainStoreTestCase(HomeserverTestCase):
 
         for e in events:
             e.internal_metadata.stream_ordering = self._next_stream_ordering
+            e.internal_metadata.instance_name = self.hs.get_instance_name()
             self._next_stream_ordering += 1
 
         def _persist(txn: LoggingTransaction) -> None: