MSC4140: don't cancel delayed state on own state (#17810)

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

For context, see [the relevant section in the
MSC](https://github.com/matrix-org/matrix-spec-proposals/blob/a09a883d9a013ac4b6ffddebd7ea87a827d211b9/proposals/4140-delayed-events-futures.md#delayed-state-events-are-cancelled-by-a-more-recent-state-event).

Author: AndrewFerr

changelog.d/17810.feature

@@ -0,0 +1 @@
+Update MSC4140 implementation to no longer cancel a user's own delayed state events with an event type & state key that match a more recent state event sent by that user.

synapse/handlers/delayed_events.py

@@ -191,18 +191,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,25 +424,37 @@ 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.
 
+        Args:
+            room_id: The room ID to match against.
+            event_type: The event type to match against.
+            state_key: The state key to match against.
+            not_from_localpart: The localpart of a user whose delayed events to not cancel.
+                If set to the empty string, any users' delayed events may be cancelled.
+
         Returns: The send time of the next delayed event to be sent, if any.
         """
 
         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)