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 09609cb0db (diff-5cd773fb307aa754bd3948871ba118b1ef0303f4d72d42a2d21e38242bf4e096R405-R410)
This commit is contained in:
Eric Eastwood 2024-06-13 11:32:50 -05:00 committed by GitHub
parent ebdce69f6a
commit 8c58eb7f17
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 31 additions and 9 deletions

1
changelog.d/17300.misc Normal file
View File

@ -0,0 +1 @@
Expose the worker instance that persisted the event on `event.internal_metadata.instance_name`.

View File

@ -204,6 +204,8 @@ pub struct EventInternalMetadata {
/// The stream ordering of this event. None, until it has been persisted. /// The stream ordering of this event. None, until it has been persisted.
#[pyo3(get, set)] #[pyo3(get, set)]
stream_ordering: Option<NonZeroI64>, stream_ordering: Option<NonZeroI64>,
#[pyo3(get, set)]
instance_name: Option<String>,
/// whether this event is an outlier (ie, whether we have the state at that /// whether this event is an outlier (ie, whether we have the state at that
/// point in the DAG) /// point in the DAG)
@ -232,6 +234,7 @@ impl EventInternalMetadata {
Ok(EventInternalMetadata { Ok(EventInternalMetadata {
data, data,
stream_ordering: None, stream_ordering: None,
instance_name: None,
outlier: false, outlier: false,
}) })
} }

View File

@ -90,6 +90,7 @@ def prune_event(event: EventBase) -> EventBase:
pruned_event.internal_metadata.stream_ordering = ( pruned_event.internal_metadata.stream_ordering = (
event.internal_metadata.stream_ordering event.internal_metadata.stream_ordering
) )
pruned_event.internal_metadata.instance_name = event.internal_metadata.instance_name
pruned_event.internal_metadata.outlier = event.internal_metadata.outlier pruned_event.internal_metadata.outlier = event.internal_metadata.outlier
# Mark the event as redacted # Mark the event as redacted
@ -116,6 +117,7 @@ def clone_event(event: EventBase) -> EventBase:
new_event.internal_metadata.stream_ordering = ( new_event.internal_metadata.stream_ordering = (
event.internal_metadata.stream_ordering event.internal_metadata.stream_ordering
) )
new_event.internal_metadata.instance_name = event.internal_metadata.instance_name
new_event.internal_metadata.outlier = event.internal_metadata.outlier new_event.internal_metadata.outlier = event.internal_metadata.outlier
return new_event return new_event

View File

@ -1551,6 +1551,7 @@ class EventCreationHandler:
# stream_ordering entry manually (as it was persisted on # stream_ordering entry manually (as it was persisted on
# another worker). # another worker).
event.internal_metadata.stream_ordering = stream_id event.internal_metadata.stream_ordering = stream_id
event.internal_metadata.instance_name = writer_instance
return event return event

View File

@ -207,6 +207,7 @@ class PersistEventsStore:
async with stream_ordering_manager as stream_orderings: async with stream_ordering_manager as stream_orderings:
for (event, _), stream in zip(events_and_contexts, stream_orderings): for (event, _), stream in zip(events_and_contexts, stream_orderings):
event.internal_metadata.stream_ordering = stream event.internal_metadata.stream_ordering = stream
event.internal_metadata.instance_name = self._instance_name
await self.db_pool.runInteraction( await self.db_pool.runInteraction(
"persist_events", "persist_events",

View File

@ -156,6 +156,7 @@ class _EventRow:
event_id: str event_id: str
stream_ordering: int stream_ordering: int
instance_name: str
json: str json: str
internal_metadata: str internal_metadata: str
format_version: Optional[int] format_version: Optional[int]
@ -1354,6 +1355,7 @@ class EventsWorkerStore(SQLBaseStore):
rejected_reason=rejected_reason, rejected_reason=rejected_reason,
) )
original_ev.internal_metadata.stream_ordering = row.stream_ordering original_ev.internal_metadata.stream_ordering = row.stream_ordering
original_ev.internal_metadata.instance_name = row.instance_name
original_ev.internal_metadata.outlier = row.outlier original_ev.internal_metadata.outlier = row.outlier
# Consistency check: if the content of the event has been modified in the # Consistency check: if the content of the event has been modified in the
@ -1439,6 +1441,7 @@ class EventsWorkerStore(SQLBaseStore):
SELECT SELECT
e.event_id, e.event_id,
e.stream_ordering, e.stream_ordering,
e.instance_name,
ej.internal_metadata, ej.internal_metadata,
ej.json, ej.json,
ej.format_version, ej.format_version,
@ -1462,13 +1465,14 @@ class EventsWorkerStore(SQLBaseStore):
event_dict[event_id] = _EventRow( event_dict[event_id] = _EventRow(
event_id=event_id, event_id=event_id,
stream_ordering=row[1], stream_ordering=row[1],
internal_metadata=row[2], instance_name=row[2],
json=row[3], internal_metadata=row[3],
format_version=row[4], json=row[4],
room_version_id=row[5], format_version=row[5],
rejected_reason=row[6], room_version_id=row[6],
rejected_reason=row[7],
redactions=[], redactions=[],
outlier=bool(row[7]), # This is an int in SQLite3 outlier=bool(row[8]), # This is an int in SQLite3
) )
# check for redactions # check for redactions

View File

@ -19,6 +19,8 @@ class EventInternalMetadata:
stream_ordering: Optional[int] stream_ordering: Optional[int]
"""the stream ordering of this event. None, until it has been persisted.""" """the stream ordering of this event. None, until it has been persisted."""
instance_name: Optional[str]
"""the instance name of the server that persisted this event. None, until it has been persisted."""
outlier: bool outlier: bool
"""whether this event is an outlier (ie, whether we have the state at that """whether this event is an outlier (ie, whether we have the state at that

View File

@ -625,6 +625,8 @@ class CloneEventTestCase(stdlib_unittest.TestCase):
) )
original.internal_metadata.stream_ordering = 1234 original.internal_metadata.stream_ordering = 1234
self.assertEqual(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 = clone_event(original)
cloned.unsigned["b"] = 3 cloned.unsigned["b"] = 3
@ -632,6 +634,7 @@ class CloneEventTestCase(stdlib_unittest.TestCase):
self.assertEqual(original.unsigned, {"a": 1, "b": 2}) self.assertEqual(original.unsigned, {"a": 1, "b": 2})
self.assertEqual(cloned.unsigned, {"a": 1, "b": 3}) self.assertEqual(cloned.unsigned, {"a": 1, "b": 3})
self.assertEqual(cloned.internal_metadata.stream_ordering, 1234) self.assertEqual(cloned.internal_metadata.stream_ordering, 1234)
self.assertEqual(cloned.internal_metadata.instance_name, "worker1")
self.assertEqual(cloned.internal_metadata.txn_id, "txn") self.assertEqual(cloned.internal_metadata.txn_id, "txn")

View File

@ -141,6 +141,7 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
self.persist(type="m.room.create", key="", creator=USER_ID) self.persist(type="m.room.create", key="", creator=USER_ID)
self.check("get_invited_rooms_for_local_user", [USER_ID_2], []) self.check("get_invited_rooms_for_local_user", [USER_ID_2], [])
event = self.persist(type="m.room.member", key=USER_ID_2, membership="invite") 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 assert event.internal_metadata.stream_ordering is not None
self.replicate() self.replicate()
@ -155,7 +156,7 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
"invite", "invite",
event.event_id, event.event_id,
PersistedEventPosition( PersistedEventPosition(
self.hs.get_instance_name(), event.internal_metadata.instance_name,
event.internal_metadata.stream_ordering, event.internal_metadata.stream_ordering,
), ),
RoomVersions.V1.identifier, RoomVersions.V1.identifier,
@ -232,11 +233,12 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
j2 = self.persist( j2 = self.persist(
type="m.room.member", sender=USER_ID_2, key=USER_ID_2, membership="join" 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 assert j2.internal_metadata.stream_ordering is not None
self.replicate() self.replicate()
expected_pos = PersistedEventPosition( expected_pos = PersistedEventPosition(
"master", j2.internal_metadata.stream_ordering j2.internal_metadata.instance_name, j2.internal_metadata.stream_ordering
) )
self.check( self.check(
"get_rooms_for_user_with_stream_ordering", "get_rooms_for_user_with_stream_ordering",
@ -288,6 +290,7 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
msg, msgctx = self.build_event() msg, msgctx = self.build_event()
self.get_success(self.persistance.persist_events([(j2, j2ctx), (msg, msgctx)])) self.get_success(self.persistance.persist_events([(j2, j2ctx), (msg, msgctx)]))
self.replicate() self.replicate()
assert j2.internal_metadata.instance_name is not None
assert j2.internal_metadata.stream_ordering is not None assert j2.internal_metadata.stream_ordering is not None
event_source = RoomEventSource(self.hs) event_source = RoomEventSource(self.hs)
@ -329,7 +332,8 @@ class EventsWorkerStoreTestCase(BaseWorkerStoreTestCase):
# joined_rooms list. # joined_rooms list.
if membership_changes: if membership_changes:
expected_pos = PersistedEventPosition( expected_pos = PersistedEventPosition(
"master", j2.internal_metadata.stream_ordering j2.internal_metadata.instance_name,
j2.internal_metadata.stream_ordering,
) )
self.assertEqual( self.assertEqual(
joined_rooms, joined_rooms,

View File

@ -431,6 +431,7 @@ class EventChainStoreTestCase(HomeserverTestCase):
for e in events: for e in events:
e.internal_metadata.stream_ordering = self._next_stream_ordering e.internal_metadata.stream_ordering = self._next_stream_ordering
e.internal_metadata.instance_name = self.hs.get_instance_name()
self._next_stream_ordering += 1 self._next_stream_ordering += 1
def _persist(txn: LoggingTransaction) -> None: def _persist(txn: LoggingTransaction) -> None: