Use a tri-state for soft failed to communicate when we need to cache invalidate

synapse/api/constants.py

@@ -372,3 +372,10 @@ class StickyEvent:
     QUERY_PARAM_NAME: Final = "org.matrix.msc4354.sticky_duration_ms"
     FIELD_NAME: Final = "msc4354_sticky"
     MAX_DURATION_MS: Final = 3600000  # 1 hour
+
+
+# for the database
+class StickyEventSoftFailed(enum.IntEnum):
+    FALSE = 0
+    TRUE = 1
+    FORMER_TRUE = 2

synapse/replication/tcp/streams/_base.py

@@ -34,7 +34,7 @@ from typing import (
 
 import attr
 
-from synapse.api.constants import AccountDataTypes
+from synapse.api.constants import AccountDataTypes, StickyEventSoftFailed
 from synapse.replication.http.streams import ReplicationGetStreamUpdates
 
 if TYPE_CHECKING:
@@ -768,16 +768,18 @@ class ThreadSubscriptionsStream(_StreamFromIdGen):
         return rows, rows[-1][0], len(updates) == limit
 
 
+@attr.s(slots=True, auto_attribs=True)
+class StickyEventsStreamRow:
+    """Stream to inform workers about changes to sticky events."""
+
+    room_id: str
+    event_id: str  # The sticky event ID
+    soft_failed_status: StickyEventSoftFailed
+
+
 class StickyEventsStream(_StreamFromIdGen):
     """A sticky event was changed."""
 
-    @attr.s(slots=True, auto_attribs=True)
-    class StickyEventsStreamRow:
-        """Stream to inform workers about changes to sticky events."""
-
-        room_id: str
-        event_id: str  # The sticky event ID
-
     NAME = "sticky_events"
     ROW_TYPE = StickyEventsStreamRow
 
@@ -799,9 +801,9 @@ class StickyEventsStream(_StreamFromIdGen):
             (
                 stream_id,
                 # These are the args to `StickyEventsStreamRow`
-                (room_id, event_id),
+                (room_id, event_id, soft_failed),
             )
-            for stream_id, room_id, event_id in updates
+            for stream_id, room_id, event_id, soft_failed in updates
         ]
 
         if not rows:

synapse/storage/databases/main/events_worker.py

@@ -45,7 +45,7 @@ from prometheus_client import Gauge
 
 from twisted.internet import defer
 
-from synapse.api.constants import Direction, EventTypes
+from synapse.api.constants import Direction, EventTypes, StickyEventSoftFailed
 from synapse.api.errors import NotFoundError, SynapseError
 from synapse.api.room_versions import (
     KNOWN_ROOM_VERSIONS,
@@ -74,6 +74,10 @@ from synapse.metrics.background_process_metrics import (
     wrap_as_background_process,
 )
 from synapse.replication.tcp.streams import BackfillStream, UnPartialStatedEventStream
+from synapse.replication.tcp.streams._base import (
+    StickyEventsStream,
+    StickyEventsStreamRow,
+)
 from synapse.replication.tcp.streams.events import EventsStream
 from synapse.replication.tcp.streams.partial_state import UnPartialStatedEventStreamRow
 from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
@@ -463,6 +467,12 @@ class EventsWorkerStore(SQLBaseStore):
                     # If the partial-stated event became rejected or unrejected
                     # when it wasn't before, we need to invalidate this cache.
                     self._invalidate_local_get_event_cache(row.event_id)
+        elif stream_name == StickyEventsStream.NAME:
+            for row in rows:
+                assert isinstance(row, StickyEventsStreamRow)
+                if row.soft_failed_status == StickyEventSoftFailed.FORMER_TRUE:
+                    # was soft-failed, now not, so invalidate caches
+                    self._invalidate_local_get_event_cache(row.event_id)
 
         super().process_replication_rows(stream_name, instance_name, token, rows)
 

synapse/storage/databases/main/sticky_events.py

@@ -28,7 +28,7 @@ from typing import (
 from twisted.internet.defer import Deferred
 
 from synapse import event_auth
-from synapse.api.constants import EventTypes, StickyEvent
+from synapse.api.constants import EventTypes, StickyEvent, StickyEventSoftFailed
 from synapse.api.errors import AuthError
 from synapse.events import EventBase
 from synapse.events.snapshot import EventPersistencePair
@@ -171,15 +171,15 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
         txn.execute(
             f"""
             SELECT stream_id, room_id, event_id FROM sticky_events
-            WHERE soft_failed=FALSE AND expires_at > ? AND stream_id > ? AND stream_id <= ? AND {clause}
+            WHERE soft_failed != ? AND expires_at > ? AND stream_id > ? AND stream_id <= ? AND {clause}
             """,
-            (now, from_id, to_id, *room_id_values),
+            (StickyEventSoftFailed.TRUE, now, from_id, to_id, *room_id_values),
         )
         return cast(List[Tuple[int, str, str]], txn.fetchall())
 
     async def get_updated_sticky_events(
         self, from_id: int, to_id: int, limit: int
-    ) -> List[Tuple[int, str, str]]:
+    ) -> List[Tuple[int, str, str, StickyEventSoftFailed]]:
         """Get updates to sticky events between two stream IDs.
 
         Args:
@@ -200,14 +200,14 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
 
     def _get_updated_sticky_events_txn(
         self, txn: LoggingTransaction, from_id: int, to_id: int, limit: int
-    ) -> List[Tuple[int, str, str]]:
+    ) -> List[Tuple[int, str, str, StickyEventSoftFailed]]:
         txn.execute(
             """
-            SELECT stream_id, room_id, event_id FROM sticky_events WHERE stream_id > ? AND stream_id <= ? LIMIT ?
+            SELECT stream_id, room_id, event_id, soft_failed FROM sticky_events WHERE stream_id > ? AND stream_id <= ? LIMIT ?
             """,
             (from_id, to_id, limit),
         )
-        return cast(List[Tuple[int, str, str]], txn.fetchall())
+        return cast(List[Tuple[int, str, str, StickyEventSoftFailed]], txn.fetchall())
 
     async def get_sticky_event_ids_sent_by_self(
         self, room_id: str, from_stream_pos: int
@@ -237,9 +237,9 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
             """
             SELECT sticky_events.event_id, sticky_events.sender, events.stream_ordering FROM sticky_events
             INNER JOIN events ON events.event_id = sticky_events.event_id
-            WHERE soft_failed=FALSE AND expires_at > ? AND sticky_events.room_id = ?
+            WHERE soft_failed=? AND expires_at > ? AND sticky_events.room_id = ?
             """,
-            (now_ms, room_id),
+            (StickyEventSoftFailed.FALSE, now_ms, room_id),
         )
         rows = cast(List[Tuple[str, str, int]], txn.fetchall())
         return [
@@ -341,7 +341,9 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
                     ev.event_id,
                     ev.sender,
                     expires_at,
-                    ev.internal_metadata.is_soft_failed(),
+                    StickyEventSoftFailed.TRUE
+                    if ev.internal_metadata.is_soft_failed()
+                    else StickyEventSoftFailed.FALSE,
                 )
                 for (ev, expires_at, stream_id) in sticky_events
             ],
@@ -426,7 +428,7 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
                 iterable=new_membership_changes,
                 keyvalues={
                     "room_id": room_id,
-                    "soft_failed": True,
+                    "soft_failed": StickyEventSoftFailed.TRUE,
                 },
                 retcols=("event_id",),
                 desc="_get_soft_failed_sticky_events_to_recheck_members",
@@ -457,7 +459,7 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
             table="sticky_events",
             keyvalues={
                 "room_id": room_id,
-                "soft_failed": True,
+                "soft_failed": StickyEventSoftFailed.TRUE,
             },
             retcols=("event_id",),
             desc="_get_soft_failed_sticky_events_to_recheck",
@@ -540,14 +542,14 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
                 f"""
                 UPDATE sticky_events AS se
                 SET
-                    soft_failed = FALSE,
+                    soft_failed = ?,
                     stream_id   = v.stream_id
                 FROM (VALUES
                     {values_placeholders}
                 ) AS v(event_id, stream_id)
                 WHERE se.event_id = v.event_id;
                 """,
-                params,
+                [StickyEventSoftFailed.FORMER_TRUE] + params,
             )
             # Also update the internal metadata on the event itself, so when we filter_events_for_client
             # we don't filter them out. It's a bit sad internal_metadata is TEXT and not JSONB...
@@ -574,14 +576,16 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
                 f"""
                 UPDATE sticky_events
                 SET
-                    soft_failed = FALSE,
+                    soft_failed = ?,
                     stream_id = CASE event_id
                         {case_expr}
                         ELSE stream_id
                     END
                 WHERE event_id IN ({",".join("?" * len(new_stream_ids))});
                 """,
-                params + [eid for eid, _ in new_stream_ids],
+                [StickyEventSoftFailed.FORMER_TRUE]
+                + params
+                + [eid for eid, _ in new_stream_ids],
             )
             clause, args = make_in_list_sql_clause(
                 txn.database_engine,

synapse/storage/schema/main/delta/93/01_sticky_events.sql

@@ -18,7 +18,11 @@ CREATE TABLE IF NOT EXISTS sticky_events(
   event_id TEXT NOT NULL,
   sender TEXT NOT NULL,
   expires_at BIGINT NOT NULL,
-  soft_failed BOOLEAN NOT NULL
+  -- 0=False, 1=True, 2=False-but-was-True
+  -- We need '2' to handle cache invalidation downstream.
+  -- Receiving a sticky event replication row with '2' will cause get_event
+  -- caches to be invalidated, so the soft-failure status can change.
+  soft_failed SMALLINT NOT NULL
 );
 
 -- for pulling out soft failed events by room

synapse/storage/schema/main/delta/93/01_sticky_events_seq.sql.postgres

@@ -15,4 +15,4 @@ CREATE SEQUENCE sticky_events_sequence;
 -- Synapse streams start at 2, because the default position is 1
 -- so any item inserted at position 1 is ignored.
 -- We have to use nextval not START WITH 2, see https://github.com/element-hq/synapse/issues/18712
-SELECT nextval('thread_subscriptions_sequence');
+SELECT nextval('sticky_events_sequence');