Add received_ts to the in-memory event internal metadata struct. (#20312)

Motivated by work on MSC4354, because the sticky events MSC specs that
`received_ts` is used to calculate the
start of an event's stickiness. (It would therefore be convenient to
have it available as part of event metadata, so an event can expose a
cheap helper method to check if it is sticky.)

This commit adds `received_ts` into the `event.internal_metadata` struct,
alongside `stream_ordering`.

I conceptually treat it as a sibling to `stream_ordering` and populate
it in all the same places where `stream_ordering` is populated.

Because this needs `received_ts` to be added to the HTTP replication
response for event persistence,
I also add a temporary compatibility measure that keeps rolling worker
upgrades working during this
transition.
I intend to remove the compatibility code in the next release.


I have manually tested that
```
SYNAPSE_POSTGRES=1 SYNAPSE_POSTGRES_USER=postgres SYNAPSE_POSTGRES_HOST=localhost SYNAPSE_POSTGRES_PASSWORD=postgres poetry run trial -j 6 tests.replication.test_sharded_event_persister
```
fails only when both the compatibility code is removed and the
`"received_ts"` key is commented out in the response (to omit it).
The compatibility code causes the test to pass even when `"received_ts"`
is commented out.

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>

changelog.d/20312.misc

@@ -0,0 +1 @@
+Add `received_ts` to the in-memory event internal metadata struct.

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<NonZeroI64>,
+
+    /// 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<i64>,
+
     pub instance_name: Option<String>,
 
     /// 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<Option<i64>> {
+        Ok(self.read_inner()?.received_ts)
+    }
+    #[setter]
+    fn set_received_ts(&self, val: Option<i64>) -> PyResult<()> {
+        self.write_inner()?.received_ts = val;
+        Ok(())
+    }
+
     #[getter]
     fn get_instance_name(&self) -> PyResult<Option<String>> {
         Ok(self.read_inner()?.instance_name.clone())

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

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,
             },
         )

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(),

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

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

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")
 

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