Add query parameter ts to allow appservices set the origin_server_ts for state events

changelog.d/11866.feature

@@ -0,0 +1 @@
+Allow application services to set the `origin_server_ts` of a state event by providing the query parameter `ts` in `PUT /_matrix/client/r0/rooms/{roomId}/state/{eventType}/{stateKey}`, per [MSC3316](https://github.com/matrix-org/matrix-doc/pull/3316). Contributed by @lukasdenk.

synapse/handlers/room_member.py

@@ -278,6 +278,7 @@ async def _local_membership_update(
         require_consent: bool = True,
         outlier: bool = False,
         historical: bool = False,
+        origin_server_ts: Optional[int] = None,
     ) -> Tuple[str, int]:
         """
         Internal membership update function to get an existing event or create
@@ -310,6 +311,8 @@ async def _local_membership_update(
             historical: Indicates whether the message is being inserted
                 back in time around some existing events. This is used to skip
                 a few checks and mark the event as backfilled.
+            origin_server_ts: The origin_server_ts to use if a new event is created. Uses
+                the current timestamp if set to None.
 
         Returns:
             Tuple of event ID and stream ordering position
@@ -348,6 +351,7 @@ async def _local_membership_update(
                 "state_key": user_id,
                 # For backwards compatibility:
                 "membership": membership,
+                "origin_server_ts": origin_server_ts,
             },
             txn_id=txn_id,
             allow_no_prev_events=allow_no_prev_events,
@@ -456,6 +460,7 @@ async def update_membership(
         allow_no_prev_events: bool = False,
         prev_event_ids: Optional[List[str]] = None,
         auth_event_ids: Optional[List[str]] = None,
+        origin_server_ts: Optional[int] = None,
     ) -> Tuple[str, int]:
         """Update a user's membership in a room.
 
@@ -487,6 +492,8 @@ async def update_membership(
                 The event ids to use as the auth_events for the new event.
                 Should normally be left as None, which will cause them to be calculated
                 based on the room state at the prev_events.
+            origin_server_ts: The origin_server_ts to use if a new event is created. Uses
+                the current timestamp if set to None.
 
         Returns:
             A tuple of the new event ID and stream ID.
@@ -526,6 +533,7 @@ async def update_membership(
                     allow_no_prev_events=allow_no_prev_events,
                     prev_event_ids=prev_event_ids,
                     auth_event_ids=auth_event_ids,
+                    origin_server_ts=origin_server_ts,
                 )
 
         return result
@@ -548,6 +556,7 @@ async def update_membership_locked(
         allow_no_prev_events: bool = False,
         prev_event_ids: Optional[List[str]] = None,
         auth_event_ids: Optional[List[str]] = None,
+        origin_server_ts: Optional[int] = None,
     ) -> Tuple[str, int]:
         """Helper for update_membership.
 
@@ -581,6 +590,8 @@ async def update_membership_locked(
                 The event ids to use as the auth_events for the new event.
                 Should normally be left as None, which will cause them to be calculated
                 based on the room state at the prev_events.
+            origin_server_ts: The origin_server_ts to use if a new event is created. Uses
+                the current timestamp if set to None.
 
         Returns:
             A tuple of the new event ID and stream ID.
@@ -712,6 +723,7 @@ async def update_membership_locked(
                 require_consent=require_consent,
                 outlier=outlier,
                 historical=historical,
+                origin_server_ts=origin_server_ts,
             )
 
         latest_event_ids = await self.store.get_prev_events_for_room(room_id)
@@ -935,6 +947,7 @@ async def update_membership_locked(
             content=content,
             require_consent=require_consent,
             outlier=outlier,
+            origin_server_ts=origin_server_ts,
         )
 
     async def _should_perform_remote_join(

synapse/rest/client/room.py

@@ -194,7 +194,7 @@ async def on_PUT(
 
         content = parse_json_object_from_request(request)
 
-        event_dict = {
+        event_dict: JsonDict = {
             "type": event_type,
             "content": content,
             "room_id": room_id,
@@ -204,6 +204,13 @@ async def on_PUT(
         if state_key is not None:
             event_dict["state_key"] = state_key
 
+        # Twisted will have processed the args by now.
+        assert request.args is not None
+        if requester.app_service:
+            event_dict["origin_server_ts"] = parse_integer(
+                request, "org.matrix.msc3316.ts"
+            )
+
         try:
             if event_type == EventTypes.Member:
                 membership = content.get("membership", None)
@@ -213,6 +220,7 @@ async def on_PUT(
                     room_id=room_id,
                     action=membership,
                     content=content,
+                    origin_server_ts=event_dict.get("origin_server_ts"),
                 )
             else:
                 (
@@ -261,8 +269,13 @@ async def on_POST(
 
         # Twisted will have processed the args by now.
         assert request.args is not None
-        if b"ts" in request.args and requester.app_service:
-            event_dict["origin_server_ts"] = parse_integer(request, "ts", 0)
+        if requester.app_service:
+            if b"ts" in request.args:
+                event_dict["origin_server_ts"] = parse_integer(request, "ts")
+            elif b"org.matrix.msc3316.ts" in request.args:
+                event_dict["origin_server_ts"] = parse_integer(
+                    request, "org.matrix.msc3316.ts"
+                )
 
         try:
             (

tests/rest/client/test_rooms.py

@@ -19,10 +19,14 @@
 
 import json
 from typing import Iterable, List
-from unittest.mock import Mock, call
+
+from authlib.common.urls import url_encode
+from unittest.mock import Mock, call, patch
 from urllib import parse as urlparse
 
 from twisted.internet import defer
+from twisted.internet.task import Clock
+from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
 from synapse.api.constants import (
@@ -32,13 +36,15 @@
     RelationTypes,
 )
 from synapse.api.errors import Codes, HttpResponseException
+from synapse.appservice import ApplicationService
 from synapse.handlers.pagination import PurgeStatus
 from synapse.rest import admin
-from synapse.rest.client import account, directory, login, profile, room, sync
+from synapse.rest.client import account, directory, login, profile, register, room, sync
+from synapse.server import HomeServer
 from synapse.types import JsonDict, RoomAlias, UserID, create_requester
 from synapse.util.stringutils import random_string
 
-from tests import unittest
+from tests import unittest, server
 from tests.test_utils import make_awaitable
 
 PATH_PREFIX = b"/_matrix/client/api/v1"
@@ -966,6 +972,67 @@ async def user_may_join_room(
         self.helper.join(self.room3, self.user2, expect_code=403, tok=self.tok2)
 
 
+class RoomAppserviceTsParamTestCase(unittest.HomeserverTestCase):
+    servlets = [
+        room.register_servlets,
+        synapse.rest.admin.register_servlets,
+        register.register_servlets,
+    ]
+
+    def prepare(self, reactor, clock, homeserver):
+        self.appservice_user, _ = self.register_appservice_user(
+            "as_user_potato", self.appservice.token
+        )
+
+        self.room = self.helper.create_room_as(
+            room_creator=self.appservice_user,
+            tok=self.appservice.token,
+            appservice_user_id=self.appservice_user,
+        )
+
+    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
+        config = self.default_config()
+
+        self.appservice = ApplicationService(
+            token="i_am_an_app_service",
+            hostname="test",
+            id="1234",
+            namespaces={"users": [{"regex": r"@as_user.*", "exclusive": True}]},
+            # Note: this user does not have to match the regex above
+            sender="@as_main:test",
+        )
+
+        mock_load_appservices = Mock(return_value=[self.appservice])
+        with patch(
+            "synapse.storage.databases.main.appservice.load_appservices",
+            mock_load_appservices,
+        ):
+            hs = self.setup_test_homeserver(config=config)
+        return hs
+
+
+    def test_send_event_ts(self):
+        ts = 1
+        event_id = self._send(ts)
+        # check that the event was successfully persisted to the database with the correct timestamp.
+        res = self.get_success(self.hs.get_datastore().get_event(event_id))
+        self.assertEquals(ts, res.origin_server_ts)
+
+    def _send(self, ts:int) ->str:
+        url_params = {
+            "user_id": self.appservice_user,
+            "ts": ts,
+        }
+        channel = server.make_request(
+            "PUT",
+            path=f"/_matrix/client/r0/rooms/{self.room}/send/m.room.message/{None}?" + url_encode(url_params),
+            content={"membership": "invite"},
+            access_token=self.appservice.token,
+        )
+        self.assertEqual(channel.code, 200)
+        return channel.json_body["event_id"]
+
+
 class RoomJoinRatelimitTestCase(RoomBase):
     user_id = "@sid1:red"