MSC4140: put delay_id in unsigned data for sender

changelog.d/19479.feature

@@ -0,0 +1 @@
+MSC4140: When persisting a delayed event to the timeline, include its `delay_id` in the event's `unsigned` section in `/sync` responses to the event sender.

rust/src/events/internal_metadata.rs

@@ -57,6 +57,7 @@ enum EventInternalMetadataData {
     PolicyServerSpammy(bool),
     Redacted(bool),
     TxnId(Box<str>),
+    DelayId(Box<str>),
     TokenId(i64),
     DeviceId(Box<str>),
 }
@@ -115,6 +116,10 @@ impl EventInternalMetadataData {
                 pyo3::intern!(py, "txn_id"),
                 o.into_pyobject(py).unwrap_infallible().into_any(),
             ),
+            EventInternalMetadataData::DelayId(o) => (
+                pyo3::intern!(py, "delay_id"),
+                o.into_pyobject(py).unwrap_infallible().into_any(),
+            ),
             EventInternalMetadataData::TokenId(o) => (
                 pyo3::intern!(py, "token_id"),
                 o.into_pyobject(py).unwrap_infallible().into_any(),
@@ -179,6 +184,12 @@ impl EventInternalMetadataData {
                     .map(String::into_boxed_str)
                     .with_context(|| format!("'{key_str}' has invalid type"))?,
             ),
+            "delay_id" => EventInternalMetadataData::DelayId(
+                value
+                    .extract()
+                    .map(String::into_boxed_str)
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
             "token_id" => EventInternalMetadataData::TokenId(
                 value
                     .extract()
@@ -472,6 +483,17 @@ impl EventInternalMetadata {
         set_property!(self, TxnId, obj.into_boxed_str());
     }
 
+    /// The delay ID, set only if the event was a delayed event.
+    #[getter]
+    fn get_delay_id(&self) -> PyResult<&str> {
+        let s = get_property!(self, DelayId)?;
+        Ok(s)
+    }
+    #[setter]
+    fn set_delay_id(&mut self, obj: String) {
+        set_property!(self, DelayId, obj.into_boxed_str());
+    }
+
     /// The access token ID of the user who sent this event, if any.
     #[getter]
     fn get_token_id(&self) -> PyResult<i64> {

synapse/events/utils.py

@@ -412,7 +412,7 @@ class SerializeEventConfig:
     # Function to convert from federation format to client format
     event_format: Callable[[JsonDict], JsonDict] = format_event_for_client_v1
     # The entity that requested the event. This is used to determine whether to include
-    # the transaction_id in the unsigned section of the event.
+    # the transaction_id and delay_id in the unsigned section of the event.
     requester: Requester | None = None
     # List of event fields to include. If empty, all fields will be returned.
     only_event_fields: list[str] | None = None
@@ -475,12 +475,13 @@ def serialize_event(
             config=config,
         )
 
-    # If we have a txn_id saved in the internal_metadata, we should include it in the
-    # unsigned section of the event if it was sent by the same session as the one
+    # If we have applicable fields saved in the internal_metadata, include them in the
+    # unsigned section of the event if the event was sent by the same session as the one
     # requesting the event.
     txn_id: str | None = getattr(e.internal_metadata, "txn_id", None)
+    delay_id: str | None = getattr(e.internal_metadata, "delay_id", None)
     if (
-        txn_id is not None
+        (txn_id is not None or delay_id is not None)
         and config.requester is not None
         and config.requester.user.to_string() == e.sender
     ):
@@ -491,7 +492,10 @@ def serialize_event(
         event_device_id: str | None = getattr(e.internal_metadata, "device_id", None)
         if event_device_id is not None:
             if event_device_id == config.requester.device_id:
-                d["unsigned"]["transaction_id"] = txn_id
+                if txn_id is not None:
+                    d["unsigned"]["transaction_id"] = txn_id
+                if delay_id is not None:
+                    d["unsigned"]["delay_id"] = delay_id
 
         else:
             # Fallback behaviour: only include the transaction ID if the event
@@ -512,7 +516,10 @@ def serialize_event(
                 or config.requester.is_guest
                 or config.requester.app_service
             ):
-                d["unsigned"]["transaction_id"] = txn_id
+                if txn_id is not None:
+                    d["unsigned"]["transaction_id"] = txn_id
+                if delay_id is not None:
+                    d["unsigned"]["delay_id"] = delay_id
 
     # invite_room_state and knock_room_state are a list of stripped room state events
     # that are meant to provide metadata about a room to an invitee/knocker. They are

synapse/handlers/delayed_events.py

@@ -585,6 +585,7 @@ async def _send_event(
                     requester,
                     event_dict,
                     txn_id=txn_id,
+                    delay_id=event.delay_id,
                 )
                 event_id = sent_event.event_id
         except ShadowBanError:

synapse/handlers/message.py

@@ -585,6 +585,7 @@ async def create_event(
         state_map: StateMap[str] | None = None,
         for_batch: bool = False,
         current_state_group: int | None = None,
+        delay_id: str | None = None,
     ) -> tuple[EventBase, UnpersistedEventContextBase]:
         """
         Given a dict from a client, create a new event. If bool for_batch is true, will
@@ -600,7 +601,7 @@ async def create_event(
         Args:
             requester
             event_dict: An entire event
-            txn_id
+            txn_id: The transaction ID.
             prev_event_ids:
                 the forward extremities to use as the prev_events for the
                 new event.
@@ -639,6 +640,8 @@ async def create_event(
             current_state_group: the current state group, used only for creating events for
                 batch persisting
 
+            delay_id: The delay ID of this event, if it was a delayed event.
+
         Raises:
             ResourceLimitError if server is blocked to some resource being
             exceeded
@@ -726,6 +729,9 @@ async def create_event(
         if txn_id is not None:
             builder.internal_metadata.txn_id = txn_id
 
+        if delay_id is not None:
+            builder.internal_metadata.delay_id = delay_id
+
         builder.internal_metadata.outlier = outlier
 
         event, unpersisted_context = await self.create_new_client_event(
@@ -966,6 +972,7 @@ async def create_and_send_nonmember_event(
         ignore_shadow_ban: bool = False,
         outlier: bool = False,
         depth: int | None = None,
+        delay_id: str | None = None,
     ) -> tuple[EventBase, int]:
         """
         Creates an event, then sends it.
@@ -994,6 +1001,7 @@ async def create_and_send_nonmember_event(
             depth: Override the depth used to order the event in the DAG.
                 Should normally be set to None, which will cause the depth to be calculated
                 based on the prev_events.
+            delay_id: The delay ID of this event, if it was a delayed event.
 
         Returns:
             The event, and its stream ordering (if deduplication happened,
@@ -1090,6 +1098,7 @@ async def create_and_send_nonmember_event(
                 ignore_shadow_ban=ignore_shadow_ban,
                 outlier=outlier,
                 depth=depth,
+                delay_id=delay_id,
             )
 
     async def _create_and_send_nonmember_event_locked(
@@ -1103,6 +1112,7 @@ async def _create_and_send_nonmember_event_locked(
         ignore_shadow_ban: bool = False,
         outlier: bool = False,
         depth: int | None = None,
+        delay_id: str | None = None,
     ) -> tuple[EventBase, int]:
         room_id = event_dict["room_id"]
 
@@ -1131,6 +1141,7 @@ async def _create_and_send_nonmember_event_locked(
                     state_event_ids=state_event_ids,
                     outlier=outlier,
                     depth=depth,
+                    delay_id=delay_id,
                 )
                 context = await unpersisted_context.persist(event)
 

synapse/synapse_rust/events.pyi

@@ -38,6 +38,8 @@ class EventInternalMetadata:
 
     txn_id: str
     """The transaction ID, if it was set when the event was created."""
+    delay_id: str
+    """The delay ID, set only if the event was a delayed event."""
     token_id: int
     """The access token ID of the user who sent this event, if any."""
     device_id: str

tests/rest/client/test_delayed_events.py

@@ -22,7 +22,7 @@
 
 from synapse.api.errors import Codes
 from synapse.rest import admin
-from synapse.rest.client import delayed_events, login, room, versions
+from synapse.rest.client import delayed_events, login, room, sync, versions
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util.clock import Clock
@@ -59,6 +59,7 @@ class DelayedEventsTestCase(HomeserverTestCase):
         delayed_events.register_servlets,
         login.register_servlets,
         room.register_servlets,
+        sync.register_servlets,
     ]
 
     def default_config(self) -> JsonDict:
@@ -106,6 +107,9 @@ def test_delayed_state_events_are_sent_on_timeout(self) -> None:
             self.user1_access_token,
         )
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+        delay_id = channel.json_body.get("delay_id")
+        assert delay_id is not None
+
         events = self._get_delayed_events()
         self.assertEqual(1, len(events), events)
         content = self._get_delayed_event_content(events[0])
@@ -128,6 +132,9 @@ def test_delayed_state_events_are_sent_on_timeout(self) -> None:
         )
         self.assertEqual(setter_expected, content.get(setter_key), content)
 
+        self._find_sent_delayed_event(self.user1_access_token, delay_id, True)
+        self._find_sent_delayed_event(self.user2_access_token, delay_id, False)
+
     def test_get_delayed_events_auth(self) -> None:
         channel = self.make_request("GET", PATH_PREFIX)
         self.assertEqual(HTTPStatus.UNAUTHORIZED, channel.code, channel.result)
@@ -254,6 +261,9 @@ def test_cancel_delayed_state_event(self, action_in_path: bool) -> None:
             expect_code=HTTPStatus.NOT_FOUND,
         )
 
+        self._find_sent_delayed_event(self.user1_access_token, delay_id, False)
+        self._find_sent_delayed_event(self.user2_access_token, delay_id, False)
+
     @parameterized.expand((True, False))
     @unittest.override_config(
         {"rc_delayed_event_mgmt": {"per_second": 0.5, "burst_count": 1}}
@@ -327,6 +337,9 @@ def test_send_delayed_state_event(
         )
         self.assertEqual(content_value, content.get(content_property_name), content)
 
+        self._find_sent_delayed_event(self.user1_access_token, delay_id, True)
+        self._find_sent_delayed_event(self.user2_access_token, delay_id, False)
+
     @parameterized.expand((True, False))
     @unittest.override_config({"rc_message": {"per_second": 2.5, "burst_count": 3}})
     def test_send_delayed_event_ratelimit(self, action_in_path: bool) -> None:
@@ -406,6 +419,9 @@ def test_restart_delayed_state_event(self, action_in_path: bool) -> None:
         )
         self.assertEqual(setter_expected, content.get(setter_key), content)
 
+        self._find_sent_delayed_event(self.user1_access_token, delay_id, True)
+        self._find_sent_delayed_event(self.user2_access_token, delay_id, False)
+
     @parameterized.expand((True, False))
     @unittest.override_config(
         {"rc_delayed_event_mgmt": {"per_second": 0.5, "burst_count": 1}}
@@ -450,6 +466,8 @@ def test_delayed_state_is_not_cancelled_by_new_state_from_same_user(
             self.user1_access_token,
         )
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+        delay_id = channel.json_body.get("delay_id")
+        assert delay_id is not None
         events = self._get_delayed_events()
         self.assertEqual(1, len(events), events)
 
@@ -474,6 +492,9 @@ def test_delayed_state_is_not_cancelled_by_new_state_from_same_user(
         )
         self.assertEqual(setter_expected, content.get(setter_key), content)
 
+        self._find_sent_delayed_event(self.user1_access_token, delay_id, True)
+        self._find_sent_delayed_event(self.user2_access_token, delay_id, False)
+
     def test_delayed_state_is_cancelled_by_new_state_from_other_user(
         self,
     ) -> None:
@@ -489,6 +510,8 @@ def test_delayed_state_is_cancelled_by_new_state_from_other_user(
             self.user1_access_token,
         )
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+        delay_id = channel.json_body.get("delay_id")
+        assert delay_id is not None
         events = self._get_delayed_events()
         self.assertEqual(1, len(events), events)
 
@@ -513,6 +536,9 @@ def test_delayed_state_is_cancelled_by_new_state_from_other_user(
         )
         self.assertEqual(setter_expected, content.get(setter_key), content)
 
+        self._find_sent_delayed_event(self.user1_access_token, delay_id, False)
+        self._find_sent_delayed_event(self.user2_access_token, delay_id, False)
+
     def _get_delayed_events(self) -> list[JsonDict]:
         channel = self.make_request(
             "GET",
@@ -549,6 +575,26 @@ def _update_delayed_event(
             body["action"] = action
         return self.make_request("POST", path, body)
 
+    def _find_sent_delayed_event(
+        self, access_token: str, delay_id: str, should_find: bool
+    ) -> None:
+        channel = self.make_request("GET", "/sync", b"", access_token)
+        self.assertEqual(HTTPStatus.OK, channel.code)
+
+        found = False
+        events = channel.json_body["rooms"]["join"][self.room_id]["timeline"]["events"]
+        for event in events:
+            if event["unsigned"].get("delay_id") == delay_id:
+                if not should_find:
+                    self.fail(
+                        "Found event with matching delay_id, but expected to not find one"
+                    )
+                if found:
+                    self.fail("Found multiple events with matching delay_id")
+                found = True
+        if should_find and not found:
+            self.fail("Did not find event with matching delay_id")
+
 
 def _get_path_for_delayed_state(
     room_id: str, event_type: str, state_key: str, delay_ms: int