diff --git a/changelog.d/20312.misc b/changelog.d/20312.misc new file mode 100644 index 0000000000..e56408bda7 --- /dev/null +++ b/changelog.d/20312.misc @@ -0,0 +1 @@ +Add `received_ts` to the in-memory event internal metadata struct. \ No newline at end of file diff --git a/rust/src/events/internal_metadata.rs b/rust/src/events/internal_metadata.rs index 0778fbfeaa..841965db6f 100644 --- a/rust/src/events/internal_metadata.rs +++ b/rust/src/events/internal_metadata.rs @@ -21,8 +21,8 @@ //! Implements the internal metadata class attached to events. //! //! The internal metadata is a bit like a `TypedDict`, in that most of -//! it is stored as a JSON dict in the DB (the exceptions being `outlier` -//! and `stream_ordering` which have their own columns in the database). +//! it is stored as a JSON dict in the DB (the exceptions being `outlier`, +//! `received_ts` and `stream_ordering` which have their own columns in the database). //! Most events have zero, or only a few, of these keys //! set. Therefore, since we care more about memory size than performance here, //! we store these fields in a mapping. @@ -272,6 +272,14 @@ struct EventInternalMetadataInner { /// The stream ordering of this event. None, until it has been persisted. pub stream_ordering: Option, + + /// The timestamp of receipt on this server (in milliseconds since the epoch) of this event. + /// + /// None in two cases: + /// - before the event has been persisted + /// - events before Synapse v0.16.0 do not have this available + pub received_ts: Option, + pub instance_name: Option, /// The event ID of the redaction event, if this event has been redacted. @@ -567,6 +575,7 @@ impl EventInternalMetadata { inner: Arc::new(RwLock::new(EventInternalMetadataInner { data, stream_ordering: None, + received_ts: None, instance_name: None, redacted_by: None, outlier: false, @@ -646,6 +655,16 @@ impl EventInternalMetadata { Ok(()) } + #[getter] + fn get_received_ts(&self) -> PyResult> { + Ok(self.read_inner()?.received_ts) + } + #[setter] + fn set_received_ts(&self, val: Option) -> PyResult<()> { + self.write_inner()?.received_ts = val; + Ok(()) + } + #[getter] fn get_instance_name(&self) -> PyResult> { Ok(self.read_inner()?.instance_name.clone()) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 7a8866c8ed..a980678955 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -1803,7 +1803,7 @@ class EventCreationHandler: if e.code == HTTPStatus.CONFLICT: raise PartialStateConflictError() raise - stream_id = result["stream_id"] + event_id = result["event_id"] # If we batch persisted events we return the last persisted event, otherwise @@ -1815,11 +1815,18 @@ class EventCreationHandler: # been de-duplicated, so we replace the given event with the # one already persisted. event = await self.store.get_event(event_id) + elif "received_ts" not in result: + # FIXME: COMPATIBILITY (remove after one release, introduced in Synapse v1.163.0): + # We can hit this in a rolling upgrade where the event_persister worker + # hasn't been upgraded yet and so we can't populate the `received_ts`. + # Replace our copy of the event with one pulled from the database. + event = await self.store.get_event(event_id) else: # If we newly persisted the event then we need to update its - # stream_ordering entry manually (as it was persisted on - # another worker). - event.internal_metadata.stream_ordering = stream_id + # stream_ordering and received_ts entries manually + # (as it was persisted on another worker). + event.internal_metadata.stream_ordering = result["stream_id"] + event.internal_metadata.received_ts = result["received_ts"] event.internal_metadata.instance_name = writer_instance return event diff --git a/synapse/replication/http/send_events.py b/synapse/replication/http/send_events.py index e0cfb0d17c..e9db757e83 100644 --- a/synapse/replication/http/send_events.py +++ b/synapse/replication/http/send_events.py @@ -64,7 +64,7 @@ class ReplicationSendEventsRestServlet(ReplicationEndpoint): 200 OK - { "stream_id": 12345, "event_id": "$abcdef..." } + { "stream_id": 12345, "event_id": "$abcdef...", "received_ts": 424242 } Responds with a 409 when a `PartialStateConflictError` is raised due to an event context that needs to be recomputed due to the un-partial stating of a room. @@ -171,6 +171,7 @@ class ReplicationSendEventsRestServlet(ReplicationEndpoint): 200, { "stream_id": last_event.internal_metadata.stream_ordering, + "received_ts": last_event.internal_metadata.received_ts, "event_id": last_event.event_id, }, ) diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index bda599460e..f95f5d82ea 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -372,6 +372,7 @@ class PersistEventsStore: # `_update_outliers_txn()` will fix this discrepancy (always use the # `stream_ordering` from the first time it was persisted). event.internal_metadata.stream_ordering = stream + event.internal_metadata.received_ts = self._clock.time_msec() event.internal_metadata.instance_name = self._instance_name sliding_sync_table_changes = None @@ -3078,7 +3079,7 @@ class PersistEventsStore: True, # processed event.internal_metadata.is_outlier(), int(event.origin_server_ts), - self._clock.time_msec(), + event.internal_metadata.received_ts, event.sender, "url" in event.content and isinstance(event.content["url"], str), event.get_state_key(), diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py index 873d355ea1..ec76a350a6 100644 --- a/synapse/storage/databases/main/events_worker.py +++ b/synapse/storage/databases/main/events_worker.py @@ -170,6 +170,8 @@ class _EventRow: stream_ordering: stream ordering for this event + received_ts: timestamp of receipt of this event on this server, in milliseconds since the Unix epoch. + json: json-encoded event structure internal_metadata: json-encoded internal metadata dict @@ -196,6 +198,7 @@ class _EventRow: event_id: str stream_ordering: int + received_ts: int | None instance_name: str json: str internal_metadata: str @@ -1567,6 +1570,7 @@ class EventsWorkerStore(SQLBaseStore): continue original_ev.internal_metadata.stream_ordering = row.stream_ordering + original_ev.internal_metadata.received_ts = row.received_ts original_ev.internal_metadata.instance_name = row.instance_name original_ev.internal_metadata.outlier = row.outlier @@ -1654,6 +1658,7 @@ class EventsWorkerStore(SQLBaseStore): SELECT e.event_id, e.stream_ordering, + e.received_ts, e.instance_name, ej.internal_metadata, ej.json, @@ -1678,16 +1683,17 @@ class EventsWorkerStore(SQLBaseStore): event_dict[event_id] = _EventRow( event_id=event_id, stream_ordering=row[1], + received_ts=row[2], # If instance_name is null we default to "master" - instance_name=row[2] or "master", - internal_metadata=row[3], - json=row[4], - format_version=row[5], - room_version_id=row[6], - rejected_reason=row[7], + instance_name=row[3] or "master", + internal_metadata=row[4], + json=row[5], + format_version=row[6], + room_version_id=row[7], + rejected_reason=row[8], unconfirmed_redactions=[], confirmed_redactions=[], - outlier=bool(row[8]), # This is an int in SQLite3 + outlier=bool(row[9]), # This is an int in SQLite3 ) # check for redactions diff --git a/synapse/synapse_rust/events.pyi b/synapse/synapse_rust/events.pyi index eb6876b0f1..07e94b769b 100644 --- a/synapse/synapse_rust/events.pyi +++ b/synapse/synapse_rust/events.pyi @@ -21,6 +21,14 @@ class EventInternalMetadata: stream_ordering: int | None """the stream ordering of this event. None, until it has been persisted.""" + received_ts: int | None + """ + The timestamp of receipt on this server (in milliseconds since the epoch) of this event. + + None in two cases: + - before the event has been persisted + - events before Synapse v0.16.0 do not have this available. + """ instance_name: str | None """the instance name of the server that persisted this event. None, until it has been persisted.""" redacted_by: str | None diff --git a/tests/events/test_utils.py b/tests/events/test_utils.py index d018568798..c5f473186e 100644 --- a/tests/events/test_utils.py +++ b/tests/events/test_utils.py @@ -631,6 +631,8 @@ class CloneEventTestCase(stdlib_unittest.TestCase): ) original.internal_metadata.stream_ordering = 1234 self.assertEqual(original.internal_metadata.stream_ordering, 1234) + original.internal_metadata.received_ts = 424242 + self.assertEqual(original.internal_metadata.received_ts, 424242) original.internal_metadata.instance_name = "worker1" self.assertEqual(original.internal_metadata.instance_name, "worker1") @@ -644,6 +646,7 @@ class CloneEventTestCase(stdlib_unittest.TestCase): cloned.unsigned.for_event(), {"age_ts": 3, "replaces_state": "2"} ) self.assertEqual(cloned.internal_metadata.stream_ordering, 1234) + self.assertEqual(cloned.internal_metadata.received_ts, 424242) self.assertEqual(cloned.internal_metadata.instance_name, "worker1") self.assertEqual(cloned.internal_metadata.txn_id, "txn") diff --git a/tests/storage/test_event_chain.py b/tests/storage/test_event_chain.py index d09437c080..15e8f1c4b1 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.received_ts = self.hs.get_clock().time_msec() e.internal_metadata.instance_name = self.hs.get_instance_name() self._next_stream_ordering += 1