MSC4140: don't cancel delayed state on own state

When a user sends a state event, do not cancel their own delayed events
for the same piece of state.

Author: AndrewFerr

synapse/handlers/delayed_events.py

@@ -181,18 +181,36 @@ async def _unsafe_process_new_event(self) -> None:
 
     async def _handle_state_deltas(self, deltas: List[StateDelta]) -> None:
         """
-        Process current state deltas to cancel pending delayed events
+        Process current state deltas to cancel other users' pending delayed events
         that target the same state.
         """
         for delta in deltas:
+            if delta.event_id is None:
+                logger.debug(
+                    "Not handling delta for deleted state: %r %r",
+                    delta.event_type,
+                    delta.state_key,
+                )
+                continue
+
             logger.debug(
                 "Handling: %r %r, %s", delta.event_type, delta.state_key, delta.event_id
             )
 
+            event = await self._store.get_event(
+                delta.event_id, check_room_id=delta.room_id
+            )
+            sender = UserID.from_string(event.sender)
+
             next_send_ts = await self._store.cancel_delayed_state_events(
                 room_id=delta.room_id,
                 event_type=delta.event_type,
                 state_key=delta.state_key,
+                not_from_localpart=(
+                    sender.localpart
+                    if sender.domain == self._config.server.server_name
+                    else ""
+                ),
             )
 
             if self._next_send_ts_changed(next_send_ts):

synapse/storage/databases/main/delayed_events.py

@@ -424,6 +424,7 @@ async def cancel_delayed_state_events(
         room_id: str,
         event_type: str,
         state_key: str,
+        not_from_localpart: str,
     ) -> Optional[Timestamp]:
         """
         Cancels all matching delayed state events, i.e. remove them as long as they haven't been processed.
@@ -434,15 +435,19 @@ async def cancel_delayed_state_events(
         def cancel_delayed_state_events_txn(
             txn: LoggingTransaction,
         ) -> Optional[Timestamp]:
-            self.db_pool.simple_delete_txn(
-                txn,
-                table="delayed_events",
-                keyvalues={
-                    "room_id": room_id,
-                    "event_type": event_type,
-                    "state_key": state_key,
-                    "is_processed": False,
-                },
+            txn.execute(
+                """
+                DELETE FROM delayed_events
+                WHERE room_id = ? AND event_type = ? AND state_key = ?
+                    AND user_localpart <> ?
+                    AND NOT is_processed
+                """,
+                (
+                    room_id,
+                    event_type,
+                    state_key,
+                    not_from_localpart,
+                ),
             )
             return self._get_next_delayed_event_send_ts_txn(txn)