From aac3c846a8e13c22e9b3cac768075719d52d3d7f Mon Sep 17 00:00:00 2001 From: Kegan Dougal <7190048+kegsay@users.noreply.github.com> Date: Thu, 2 Oct 2025 16:47:45 +0100 Subject: [PATCH] Use a tri-state for soft failed to communicate when we need to cache invalidate --- synapse/api/constants.py | 7 ++++ synapse/replication/tcp/streams/_base.py | 22 ++++++------ .../storage/databases/main/events_worker.py | 12 ++++++- .../storage/databases/main/sticky_events.py | 36 ++++++++++--------- .../schema/main/delta/93/01_sticky_events.sql | 6 +++- .../93/01_sticky_events_seq.sql.postgres | 2 +- 6 files changed, 56 insertions(+), 29 deletions(-) diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 95ec369bc2..c5d59a8211 100644 --- a/synapse/api/constants.py +++ b/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 diff --git a/synapse/replication/tcp/streams/_base.py b/synapse/replication/tcp/streams/_base.py index bdc74e4b6a..ab3fa02a78 100644 --- a/synapse/replication/tcp/streams/_base.py +++ b/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: diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py index 31e2312211..17579cc465 100644 --- a/synapse/storage/databases/main/events_worker.py +++ b/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) diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py index 082df620bb..077518cfbb 100644 --- a/synapse/storage/databases/main/sticky_events.py +++ b/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, diff --git a/synapse/storage/schema/main/delta/93/01_sticky_events.sql b/synapse/storage/schema/main/delta/93/01_sticky_events.sql index 0c9319a7d2..18cce22fbc 100644 --- a/synapse/storage/schema/main/delta/93/01_sticky_events.sql +++ b/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 diff --git a/synapse/storage/schema/main/delta/93/01_sticky_events_seq.sql.postgres b/synapse/storage/schema/main/delta/93/01_sticky_events_seq.sql.postgres index 5a28a309d9..9ba72856bc 100644 --- a/synapse/storage/schema/main/delta/93/01_sticky_events_seq.sql.postgres +++ b/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');