Remove the stickiness of MSC4354 Sticky Events when they are redacted after persistence. (#20323)

Part of: MSC4354
Part of: #19641
Follows: #20232 (redacted before persistence), #20322 (redacted during
persistence)

Original commit schedule, with full messages:

- Add tests for redaction of Sticky Events

- Add `delete_sticky_events_txn`

- Remove stickiness from validly redacted events at persistence


### Dev notes

Overview:

- [`#20232`](https://github.com/element-hq/synapse/pull/20232) (redacted
before persistence)
- [`#20322`](https://github.com/element-hq/synapse/pull/20322) (redacted
during persistence)
- [`#20323`](https://github.com/element-hq/synapse/pull/20323) (redacted
after persistence)

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>

changelog.d/20323.bugfix

@@ -0,0 +1 @@
+Remove the stickiness of [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) when they are redacted after persistence.

synapse/storage/databases/main/events.py

@@ -81,6 +81,7 @@ from synapse.types import (
     StrCollection,
     UserID,
     get_domain_from_id,
+    unwrap,
 )
 from synapse.types.handlers import SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES
 from synapse.types.state import StateFilter
@@ -1184,6 +1185,18 @@ class PersistEventsStore:
             txn, room_id, room_version, events_and_contexts=events_and_contexts
         )
 
+        confirmed_redacted_already_persisted_event_ids = (
+            self._compute_newly_redacted_event_ids_txn(
+                txn, room_id, room_version, events_and_contexts=events_and_contexts
+            )
+        )
+        if confirmed_redacted_already_persisted_event_ids:
+            # If any of our already-existing events being redacted are sticky,
+            # we should remove the stickiness.
+            self.store.delete_sticky_events_txn(
+                txn, confirmed_redacted_already_persisted_event_ids
+            )
+
         self._store_event_txn(txn, events_and_contexts=events_and_contexts)
 
         if new_forward_extremities:
@@ -3012,6 +3025,116 @@ class PersistEventsStore:
 
         return out
 
+    def _compute_newly_redacted_event_ids_txn(
+        self,
+        txn: LoggingTransaction,
+        room_id: str,
+        room_version: RoomVersion,
+        events_and_contexts: list[EventPersistencePair],
+    ) -> set[str]:
+        """
+        If we are persisting any redactions, computes the set of already-persisted event IDs
+        that we can confirm are now redacted.
+
+        Put another way: this function tells you about _known_ events that are getting
+        redacted by the persistence of this batch of events.
+        (It's useful for applying some effects to events as they become redacted, such as
+        removing their stickiness.)
+
+        For redactions requiring re-check, applies the re-check and only returns events satisfying
+        the re-check.
+        For redactions requiring re-check, if we don't have the target event that is being redacted,
+        the event ID is NOT returned as we can't confirm it.
+
+        See:
+            `EventInternalMetadata::need_to_check_redaction` for the concept of room v3+ rechecks.
+
+        Args:
+            room_id: ID of the room that we are persisting events for
+                and that we might be redacting events within
+            room_version: Version of the room that we are persisting events for
+            events_and_contexts: The batch of events (with their persistence contexts) being persisted
+
+        Returns:
+            Set of already-persisted event IDs that are newly-redacted.
+            The events have been verified to be in the same room.
+        """
+
+        redaction_events = [
+            event
+            for event, _context in events_and_contexts
+            if event.redacts is not None and event.rejected_reason is None
+        ]
+        if not redaction_events:
+            return set()
+
+        # Fetch the events getting redacted, for 3 reasons:
+        # - to check their event type (we don't apply redactions to `m.room.create` events)
+        # - to check they are in the same room
+        # - to get the domain of their sender (needed for confirming some redactions)
+        raw_rows = self.db_pool.simple_select_many_txn(
+            txn,
+            table="events",
+            column="event_id",
+            iterable=[
+                # seen to be non-None above
+                unwrap(e.redacts)
+                for e in redaction_events
+            ],
+            # This `room_id` match is critical to ensure redactions don't try to redact events
+            # from other rooms!
+            keyvalues={"room_id": room_id},
+            retcols=("event_id", "sender", "type"),
+        )
+
+        # {event_id: (sender_domain, event_type), ...}
+        redacted_event_rows_by_event_id: dict[str, tuple[str, str]] = {}
+        for event_id, sender, event_type in raw_rows:
+            redacted_event_rows_by_event_id[event_id] = (
+                get_domain_from_id(sender),
+                event_type,
+            )
+
+        confirmed_redaction_events: list[EventBase] = []
+        for redaction_event in redaction_events:
+            # unwrap: seen to be non-None above
+            redacted_event_row = redacted_event_rows_by_event_id.get(
+                unwrap(redaction_event.redacts)
+            )
+
+            if redacted_event_row is None:
+                # We don't have the event that this redaction is redacting,
+                # so we can't confirm the redaction.
+                continue
+
+            redacted_event_sender_domain, redacted_event_type = redacted_event_row
+
+            if redacted_event_type == EventTypes.Create:
+                # we choose to ignore redactions of m.room.create events,
+                # as in `_maybe_redact_event_row`
+                continue
+
+            # Some redactions in v3+ rooms need a recheck based on the event
+            # they are redacting.
+            if redaction_event.internal_metadata.need_to_check_redaction():
+                # This is the same logic as in `_maybe_redact_event_row`.
+                if (
+                    get_domain_from_id(redaction_event.sender)
+                    != redacted_event_sender_domain
+                ):
+                    # Sender servers don't match, so the event isn't actually redacted
+                    continue
+
+                # This redaction event is allowed. Mark as not needing a recheck.
+                redaction_event.internal_metadata.recheck_redaction = False
+
+            confirmed_redaction_events.append(redaction_event)
+
+        return {
+            unwrap(redaction_event.redacts)
+            for redaction_event in confirmed_redaction_events
+        }
+
     def _store_event_txn(
         self,
         txn: LoggingTransaction,

synapse/storage/databases/main/events_worker.py

@@ -1816,7 +1816,11 @@ class EventsWorkerStore(SQLBaseStore):
             # Starting in room version v3, some redactions need to be
             # rechecked if we didn't have the redacted event at the
             # time, so we recheck on read instead.
-            # NOTE: If this logic changes, need to update `_apply_existing_redaction_txn`
+            # NOTE: If this logic changes, need to update:
+            # - `_apply_existing_redaction_txn` (persistence of an event that is already redacted)
+            # - `_compute_newly_redacted_event_ids_txn` (persistence of a new redaction)
+            # - `persist_and_notify_client_events` (sort of; applies quality-of-life errors to clients sending
+            #   redactions that wouldn't be valid if they were sent.)
             if redaction_event.internal_metadata.need_to_check_redaction():
                 expected_domain = get_domain_from_id(original_ev.sender)
                 if get_domain_from_id(redaction_event.sender) == expected_domain:

synapse/storage/databases/main/sticky_events.py

@@ -430,6 +430,38 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
             ],
         )
 
+    def delete_sticky_events_txn(
+        self, txn: LoggingTransaction, event_ids: Collection[str]
+    ) -> None:
+        """
+        Given a list of event IDs, deletes `sticky_events` entries for those events.
+        This prevents us from retrieving those events as if they are still sticky
+        in the future.
+
+        Ignores entries that are not present in the `sticky_events` table.
+
+        This is used when an event is redacted.
+        """
+        if not event_ids:
+            return
+
+        event_id_clause, event_id_args = make_in_list_sql_clause(
+            self.database_engine, "event_id", event_ids
+        )
+        txn.execute(
+            # The `sticky_events` table doesn't have an index on `event_id`,
+            # so use the `events` table to assist us, by getting the events'
+            # stream orderings, which _are_ indexed in the `sticky_events` table.
+            f"""
+            DELETE FROM sticky_events
+            WHERE event_stream_ordering IN (
+                SELECT stream_ordering FROM events
+                WHERE {event_id_clause}
+            )
+            """,
+            event_id_args,
+        )
+
     async def compute_sticky_events_to_un_soft_fail(
         self,
         room_id: str,

tests/rest/client/test_sync_sticky_events.py

@@ -31,6 +31,24 @@ from tests import unittest
 from tests.utils import USE_POSTGRES_FOR_TESTS
 
 
+@dataclass
+class SyncHelperResponse:
+    sticky_event_ids: list[str]
+    """Event IDs of events returned from the sticky section, in-order."""
+
+    next_batch: str
+    """Sync token to pass to `?since=` in order to do an incremental sync."""
+
+
+NO_TIMELINE_FILTER = json.dumps(
+    {"room": {"timeline": {"types": ["io.element.example"]}}}
+)
+"""
+A filter that excludes message events, so that the sticky events come down
+the sticky section rather than the timeline section.
+"""
+
+
 class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
     """
     Tests for oldschool (v3) /sync with sticky events (MSC4354)
@@ -62,6 +80,32 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
         # Create a room
         self.room_id = self.helper.create_room_as(self.user_id, tok=self.token)
 
+    def _do_sync(self, *, since: str | None, filter_json: str) -> SyncHelperResponse:
+        """Small helper to do a sync and get the sticky events out."""
+        and_since_param = "" if since is None else f"&since={since}"
+        channel = self.make_request(
+            "GET",
+            f"/sync?filter={quote(filter_json)}{and_since_param}",
+            access_token=self.token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        # Get sticky events from the sync response
+        sticky_events = []
+        # In this test, when no sticky events are in the response,
+        # we don't have a rooms section at all
+        if "rooms" in channel.json_body:
+            sticky_events = channel.json_body["rooms"]["join"][self.room_id][
+                "msc4354_sticky"
+            ]["events"]
+
+        return SyncHelperResponse(
+            sticky_event_ids=[
+                sticky_event["event_id"] for sticky_event in sticky_events
+            ],
+            next_batch=channel.json_body["next_batch"],
+        )
+
     def test_single_sticky_event_appears_in_initial_sync(self) -> None:
         """
         Test sending a single sticky event and then doing an initial /sync.
@@ -215,16 +259,6 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
         > out by the timeline filter, the sticky event MUST appear in sticky.events.
         > — https://github.com/matrix-org/matrix-spec-proposals/blob/4340903c15e9eab1bfb2f6a31cfa08fd535f7e7c/proposals/4354-sticky-events.md#sync-api-changes
         """
-        # A filter that excludes message events
-        filter_json = json.dumps(
-            {
-                "room": {
-                    # We only want these io.element.example events
-                    "timeline": {"types": ["io.element.example"]},
-                }
-            }
-        )
-
         # Send a sticky message event (will be filtered by our filter)
         sticky_event_id = self.helper.send_sticky_event(
             self.room_id,
@@ -245,7 +279,7 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
         # Perform initial sync with our filter
         channel = self.make_request(
             "GET",
-            f"/sync?filter={quote(filter_json)}",
+            f"/sync?filter={quote(NO_TIMELINE_FILTER)}",
             access_token=self.token,
         )
         self.assertEqual(channel.code, 200, channel.result)
@@ -283,6 +317,55 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
             [sticky_event_id],
         )
 
+    def test_redacted_sticky_event_not_in_sync(self) -> None:
+        """
+        Test that a sticky event which has since been redacted is not returned in the
+        sticky section of the sync response, since redaction strips its stickiness.
+        """
+        redacted_event_id = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "to be redacted", "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+        kept_event_id = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "to be left alone", "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+
+        # Both sticky events are visible to begin with
+        self.assertEqual(
+            set(
+                self._do_sync(
+                    since=None, filter_json=NO_TIMELINE_FILTER
+                ).sticky_event_ids
+            ),
+            {redacted_event_id, kept_event_id},
+        )
+
+        # We then redact one of them.
+        channel = self.make_request(
+            "PUT",
+            f"/_matrix/client/v3/rooms/{self.room_id}/redact/{redacted_event_id}/txn1",
+            {},
+            access_token=self.token,
+        )
+        self.assertEqual(channel.code, 200, channel.result)
+
+        # ... but the redacted one is no longer sticky
+        self.assertEqual(
+            set(
+                self._do_sync(
+                    since=None, filter_json=NO_TIMELINE_FILTER
+                ).sticky_event_ids
+            ),
+            {kept_event_id},
+        )
+
     def test_ignored_users_sticky_events(self) -> None:
         """
         Test that sticky events from ignored users are not delivered to clients.
@@ -362,10 +445,6 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
         > — https://github.com/matrix-org/matrix-spec-proposals/blob/d42301b17859d046953563dc7f046edd51789a12/proposals/4354-sticky-events.md#sync-api-changes
         """
         # A filter that excludes message events, so that the sticky events come down
-        # the sticky section rather than the timeline section.
-        filter_json = json.dumps(
-            {"room": {"timeline": {"types": ["io.element.example"]}}}
-        )
 
         sticky_event_id = self.helper.send_sticky_event(
             self.room_id,
@@ -381,7 +460,7 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
         user2_token = self.login(user2_id, "pass")
         channel = self.make_request(
             "GET",
-            f"/sync?filter={quote(filter_json)}",
+            f"/sync?filter={quote(NO_TIMELINE_FILTER)}",
             access_token=user2_token,
         )
         self.assertEqual(channel.code, HTTPStatus.OK, channel.result)
@@ -393,7 +472,7 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
         # The next (incremental) sync gives user2 the historical sticky event.
         channel = self.make_request(
             "GET",
-            f"/sync?filter={quote(filter_json)}&since={since}",
+            f"/sync?filter={quote(NO_TIMELINE_FILTER)}&since={since}",
             access_token=user2_token,
         )
         self.assertEqual(channel.code, HTTPStatus.OK, channel.result)
@@ -568,18 +647,6 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
         In this test we patch it to 3 (as sending 100 events is not very efficient).
         """
 
-        # A filter that excludes message events
-        # This is needed so that the sticky events come down the sticky section
-        # and not the timeline section, which would hamper our test.
-        filter_json = json.dumps(
-            {
-                "room": {
-                    # We only want these io.element.example events
-                    "timeline": {"types": ["io.element.example"]},
-                }
-            }
-        )
-
         # Send 8 sticky events: enough for 2 full pages and then a partial page with 2.
         sent_sticky_event_ids = []
         for i in range(8):
@@ -592,53 +659,19 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase):
             )
             sent_sticky_event_ids.append(response["event_id"])
 
-        @dataclass
-        class SyncHelperResponse:
-            sticky_event_ids: list[str]
-            """Event IDs of events returned from the sticky section, in-order."""
-
-            next_batch: str
-            """Sync token to pass to `?since=` in order to do an incremental sync."""
-
-        def _do_sync(*, since: str | None) -> SyncHelperResponse:
-            """Small helper to do a sync and get the sticky events out."""
-            and_since_param = "" if since is None else f"&since={since}"
-            channel = self.make_request(
-                "GET",
-                f"/sync?filter={quote(filter_json)}{and_since_param}",
-                access_token=self.token,
-            )
-            self.assertEqual(channel.code, 200, channel.result)
-
-            # Get sticky events from the sync response
-            sticky_events = []
-            # In this test, when no sticky events are in the response,
-            # we don't have a rooms section at all
-            if "rooms" in channel.json_body:
-                sticky_events = channel.json_body["rooms"]["join"][self.room_id][
-                    "msc4354_sticky"
-                ]["events"]
-
-            return SyncHelperResponse(
-                sticky_event_ids=[
-                    sticky_event["event_id"] for sticky_event in sticky_events
-                ],
-                next_batch=channel.json_body["next_batch"],
-            )
-
         # Perform initial sync, we should get the first 3 sticky events,
         # in order.
-        sync1 = _do_sync(since=None)
+        sync1 = self._do_sync(since=None, filter_json=NO_TIMELINE_FILTER)
         self.assertEqual(sync1.sticky_event_ids, sent_sticky_event_ids[0:3])
 
         # Now do an incremental sync and expect the next page of 3
-        sync2 = _do_sync(since=sync1.next_batch)
+        sync2 = self._do_sync(since=sync1.next_batch, filter_json=NO_TIMELINE_FILTER)
         self.assertEqual(sync2.sticky_event_ids, sent_sticky_event_ids[3:6])
 
         # Now do another incremental sync and expect the last 2
-        sync3 = _do_sync(since=sync2.next_batch)
+        sync3 = self._do_sync(since=sync2.next_batch, filter_json=NO_TIMELINE_FILTER)
         self.assertEqual(sync3.sticky_event_ids, sent_sticky_event_ids[6:8])
 
         # Finally, expect an empty incremental sync at the end
-        sync4 = _do_sync(since=sync3.next_batch)
+        sync4 = self._do_sync(since=sync3.next_batch, filter_json=NO_TIMELINE_FILTER)
         self.assertEqual(sync4.sticky_event_ids, [])

tests/storage/test_sticky_events.py

@@ -159,6 +159,54 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
             ],
         )
 
+    def test_redaction_removes_sticky_event(self) -> None:
+        """
+        Tests that redacting a sticky event causes it to be removed from the `sticky_events` table.
+        """
+
+        def _sticky_event_ids_from_table() -> list[str]:
+            """Return all event IDs in the `sticky_events` table, in sticky stream order."""
+            rows = self.get_success(
+                self.store.db_pool.simple_select_list(
+                    table="sticky_events",
+                    keyvalues=None,
+                    retcols=("stream_id", "event_id"),
+                )
+            )
+            return [event_id for _, event_id in sorted(rows)]
+
+        redacted_event_id = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "to be redacted", "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+        kept_event_id = self.helper.send_sticky_event(
+            self.room_id,
+            EventTypes.Message,
+            duration=Duration(minutes=1),
+            content={"body": "to be left alone", "msgtype": "m.text"},
+            tok=self.token,
+        )["event_id"]
+
+        # Before redaction, both sticky events are in the table.
+        self.assertEqual(
+            set(_sticky_event_ids_from_table()), {redacted_event_id, kept_event_id}
+        )
+
+        # Redact
+        channel = self.make_request(
+            "PUT",
+            f"/_matrix/client/v3/rooms/{self.room_id}/redact/{redacted_event_id}/txn1",
+            {},
+            access_token=self.token,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.result)
+
+        # After redaction, only the unredacted sticky event is left in the table.
+        self.assertEqual(set(_sticky_event_ids_from_table()), {kept_event_id})
+
     def test_get_updated_sticky_events_with_limit(self) -> None:
         """Test that the limit parameter works correctly."""
         # Get the starting stream_id