Make the event txnId mapping rely on the device_id instead of the access_token_id.

changelog.d/13083.misc

@@ -0,0 +1 @@
+Make the event `txnId` mapping rely on the `device_id` instead of the `access_token_id`.

synapse/events/__init__.py

@@ -199,7 +199,8 @@ def __init__(self, internal_metadata_dict: JsonDict):
     proactively_send: DictProperty[bool] = DictProperty("proactively_send")
     redacted: DictProperty[bool] = DictProperty("redacted")
     txn_id: DictProperty[str] = DictProperty("txn_id")
-    token_id: DictProperty[int] = DictProperty("token_id")
+    device_id: DictProperty[str] = DictProperty("device_id")
+    token_id: DictProperty[Optional[int]] = DefaultDictProperty("token_id", None)
     historical: DictProperty[bool] = DictProperty("historical")
 
     # XXX: These are set by StreamWorkerStore._set_before_and_after.

synapse/events/utils.py

@@ -317,8 +317,8 @@ class SerializeEventConfig:
     as_client_event: bool = True
     # Function to convert from federation format to client format
     event_format: Callable[[JsonDict], JsonDict] = format_event_for_client_v1
-    # ID of the user's auth token - used for namespacing of transaction IDs
-    token_id: Optional[int] = None
+    # The user's device ID which did that request - used for namespacing of transaction IDs
+    device_id: Optional[str] = None
     # List of event fields to include. If empty, all fields will be returned.
     only_event_fields: Optional[List[str]] = None
     # Some events can have stripped room state stored in the `unsigned` field.
@@ -368,11 +368,12 @@ def serialize_event(
             e.unsigned["redacted_because"], time_now_ms, config=config
         )
 
-    if config.token_id is not None:
-        if config.token_id == getattr(e.internal_metadata, "token_id", None):
-            txn_id = getattr(e.internal_metadata, "txn_id", None)
-            if txn_id is not None:
-                d["unsigned"]["transaction_id"] = txn_id
+    if config.device_id is not None and config.device_id == getattr(
+        e.internal_metadata, "device_id", None
+    ):
+        txn_id = getattr(e.internal_metadata, "txn_id", None)
+        if txn_id is not None:
+            d["unsigned"]["transaction_id"] = txn_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/message.py

@@ -698,6 +698,10 @@ async def create_event(
         if require_consent and not is_exempt:
             await self.assert_accepted_privacy_policy(requester)
 
+        if requester.device_id is not None:
+            builder.internal_metadata.device_id = requester.device_id
+
+        # We're still inserting the access_token_id in case we're rolling back
         if requester.access_token_id is not None:
             builder.internal_metadata.token_id = requester.access_token_id
 
@@ -894,12 +898,12 @@ async def get_event_from_transaction(
         Returns:
             An event if one could be found, None otherwise.
         """
-        if requester.access_token_id:
+        if requester.device_id:
             existing_event_id = await self.store.get_event_id_from_transaction_id(
-                room_id,
-                requester.user.to_string(),
-                requester.access_token_id,
-                txn_id,
+                room_id=room_id,
+                user_id=requester.user.to_string(),
+                device_id=requester.device_id,
+                txn_id=txn_id,
             )
             if existing_event_id:
                 return await self.store.get_event(existing_event_id)

synapse/handlers/room_member.py

@@ -398,12 +398,12 @@ async def _local_membership_update(
         # Check if we already have an event with a matching transaction ID. (We
         # do this check just before we persist an event as well, but may as well
         # do it up front for efficiency.)
-        if txn_id and requester.access_token_id:
+        if txn_id and requester.device_id:
             existing_event_id = await self.store.get_event_id_from_transaction_id(
-                room_id,
-                requester.user.to_string(),
-                requester.access_token_id,
-                txn_id,
+                room_id=room_id,
+                user_id=requester.user.to_string(),
+                device_id=requester.device_id,
+                txn_id=txn_id,
             )
             if existing_event_id:
                 event_pos = await self.store.get_position_for_event(existing_event_id)

synapse/rest/client/sync.py

@@ -205,7 +205,7 @@ async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
         # We know that the the requester has an access token since appservices
         # cannot use sync.
         response_content = await self.encode_response(
-            time_now, sync_result, requester.access_token_id, filter_collection
+            time_now, sync_result, requester.device_id, filter_collection
         )
 
         logger.debug("Event formatting complete")
@@ -216,7 +216,7 @@ async def encode_response(
         self,
         time_now: int,
         sync_result: SyncResult,
-        access_token_id: Optional[int],
+        device_id: Optional[str],
         filter: FilterCollection,
     ) -> JsonDict:
         logger.debug("Formatting events in sync response")
@@ -229,12 +229,12 @@ async def encode_response(
 
         serialize_options = SerializeEventConfig(
             event_format=event_formatter,
-            token_id=access_token_id,
+            device_id=device_id,
             only_event_fields=filter.event_fields,
         )
         stripped_serialize_options = SerializeEventConfig(
             event_format=event_formatter,
-            token_id=access_token_id,
+            device_id=device_id,
             include_stripped_room_state=True,
         )
 
@@ -457,13 +457,9 @@ async def encode_room(
         Args:
             room: sync result for a single room
             time_now: current time - used as a baseline for age calculations
-            token_id: ID of the user's auth token - used for namespacing
-                of transaction IDs
             joined: True if the user is joined to this room - will mean
                 we handle ephemeral events
-            only_fields: Optional. The list of event fields to include.
-            event_formatter: function to convert from federation format
-                to client format
+            serialize_options: Event serializer options
         Returns:
             The room, encoded in our response format
         """

synapse/storage/databases/main/events.py

@@ -978,14 +978,16 @@ def _persist_transaction_ids_txn(
 
         to_insert = []
         for event, _ in events_and_contexts:
+            device_id = getattr(event.internal_metadata, "device_id", None)
             token_id = getattr(event.internal_metadata, "token_id", None)
             txn_id = getattr(event.internal_metadata, "txn_id", None)
-            if token_id and txn_id:
+            if device_id and txn_id:
                 to_insert.append(
                     (
                         event.event_id,
                         event.room_id,
                         event.sender,
+                        device_id,
                         token_id,
                         txn_id,
                         self._clock.time_msec(),
@@ -1000,6 +1002,7 @@ def _persist_transaction_ids_txn(
                     "event_id",
                     "room_id",
                     "user_id",
+                    "device_id",
                     "token_id",
                     "txn_id",
                     "inserted_ts",

synapse/storage/databases/main/events_worker.py

@@ -2025,7 +2025,7 @@ def get_next_event_to_expire_txn(
         )
 
     async def get_event_id_from_transaction_id(
-        self, room_id: str, user_id: str, token_id: int, txn_id: str
+        self, room_id: str, user_id: str, device_id: str, txn_id: str
     ) -> Optional[str]:
         """Look up if we have already persisted an event for the transaction ID,
         returning the event ID if so.
@@ -2035,7 +2035,7 @@ async def get_event_id_from_transaction_id(
             keyvalues={
                 "room_id": room_id,
                 "user_id": user_id,
-                "token_id": token_id,
+                "device_id": device_id,
                 "txn_id": txn_id,
             },
             retcol="event_id",
@@ -2055,29 +2055,34 @@ async def get_already_persisted_events(
         """
 
         mapping = {}
-        txn_id_to_event: Dict[Tuple[str, int, str], str] = {}
+        txn_id_to_event: Dict[Tuple[str, str, str, str], str] = {}
 
         for event in events:
-            token_id = getattr(event.internal_metadata, "token_id", None)
             txn_id = getattr(event.internal_metadata, "txn_id", None)
+            device_id = getattr(event.internal_metadata, "device_id", None)
 
-            if token_id and txn_id:
-                # Check if this is a duplicate of an event in the given events.
-                existing = txn_id_to_event.get((event.room_id, token_id, txn_id))
+            if device_id and txn_id:
+                existing = txn_id_to_event.get(
+                    (event.room_id, event.sender, device_id, txn_id)
+                )
                 if existing:
                     mapping[event.event_id] = existing
                     continue
 
                 # Check if this is a duplicate of an event we've already
                 # persisted.
                 existing = await self.get_event_id_from_transaction_id(
-                    event.room_id, event.sender, token_id, txn_id
+                    event.room_id, event.sender, device_id, txn_id
                 )
                 if existing:
                     mapping[event.event_id] = existing
-                    txn_id_to_event[(event.room_id, token_id, txn_id)] = existing
+                    txn_id_to_event[
+                        (event.room_id, event.sender, device_id, txn_id)
+                    ] = existing
                 else:
-                    txn_id_to_event[(event.room_id, token_id, txn_id)] = event.event_id
+                    txn_id_to_event[
+                        (event.room_id, event.sender, device_id, txn_id)
+                    ] = event.event_id
 
         return mapping
 

synapse/storage/schema/main/delta/72/01event_txn_device_id.sql.postgres

@@ -0,0 +1,35 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ALTER TABLE event_txn_id
+  ADD COLUMN device_id TEXT;
+
+UPDATE event_txn_id
+  SET device_id = access_tokens.device_id
+  FROM access_tokens 
+  WHERE event_txn_id.token_id = access_tokens.id;
+
+DELETE FROM event_txn_id
+  WHERE device_id IS NULL;
+
+DROP INDEX event_txn_id_txn_id;
+CREATE UNIQUE INDEX event_txn_id_txn_id ON event_txn_id (room_id, user_id, device_id, txn_id);
+-- Keep this (non-unique) index in case we're rolling back
+CREATE INDEX event_txn_id_txn_id_token_id ON event_txn_id (room_id, user_id, token_id, txn_id);
+
+-- Make the device_id NOT NULL and remove the NOT NULL constraint from the token_id
+ALTER TABLE event_txn_id
+  ALTER COLUMN device_id SET NOT NULL,
+  ALTER COLUMN token_id DROP NOT NULL;

synapse/storage/schema/main/delta/72/01event_txn_device_id.sql.sqlite

@@ -0,0 +1,53 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE TABLE event_txn_id2 (
+  event_id TEXT NOT NULL REFERENCES events (event_id) ON DELETE CASCADE, 
+  room_id TEXT NOT NULL,
+  user_id TEXT NOT NULL,
+  device_id TEXT NOT NULL,
+  token_id TEXT, -- Keep this field in case we're rolling back
+  txn_id TEXT NOT NULL,
+  inserted_ts BIGINT NOT NULL
+);
+
+DROP INDEX event_txn_id_event_id;
+DROP INDEX event_txn_id_txn_id;
+DROP INDEX event_txn_id_ts;
+
+CREATE UNIQUE INDEX event_txn_id_event_id ON event_txn_id2(event_id);
+CREATE UNIQUE INDEX event_txn_id_txn_id ON event_txn_id2(room_id, user_id, device_id, txn_id);
+-- Keep this index in case we're rolling back
+CREATE INDEX event_txn_id_txn_id_token_id ON event_txn_id2(room_id, user_id, token_id, txn_id);
+
+INSERT INTO event_txn_id2 
+SELECT
+  event_txn_id.event_id,
+  event_txn_id.room_id,
+  event_txn_id.user_id,
+  event_txn_id.token_id,
+  access_tokens.device_id,
+  event_txn_id.txn_id,
+  event_txn_id.inserted_ts
+FROM event_txn_id
+INNER JOIN access_tokens
+  ON access_tokens.id = event_txn_id.token_id
+WHERE device_id IS NOT NULL;
+
+CREATE INDEX event_txn_id_ts ON event_txn_id2(inserted_ts);
+
+DROP TABLE event_txn_id;
+
+ALTER TABLE event_txn_id2 RENAME TO event_txn_id;

tests/handlers/test_message.py

@@ -49,15 +49,9 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.access_token = self.login("tester", "foobar")
         self.room_id = self.helper.create_room_as(self.user_id, tok=self.access_token)
 
-        info = self.get_success(
-            self.hs.get_datastores().main.get_user_by_access_token(
-                self.access_token,
-            )
+        self.requester = create_requester(
+            self.user_id, device_id="unittest-dummy-device"
         )
-        assert info is not None
-        self.token_id = info.token_id
-
-        self.requester = create_requester(self.user_id, access_token_id=self.token_id)
 
     def _create_and_persist_member_event(self) -> Tuple[EventBase, EventContext]:
         # Create a member event we can use as an auth_event