dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-09-17 08:04:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add support for un-soft-failing MSC4354 Sticky Events when room state changes, making federation support more reliable. (#20204)

Browse Source 
Part of: MSC4354

Experimental feature tracking issue:
https://github.com/element-hq/synapse/issues/19409

Related Complement tests currently in
https://github.com/matrix-org/complement/pull/806/files#diff-6c9d6d169485d0848c6b20dd9b43f6fe669a8a710e42f953d08fa25a99cc8f4cR509

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>
This commit is contained in:
Olivier 'reivilibre
2026-09-14 11:41:21 +01:00
committed by GitHub
parent 3f17bd706a
commit 6825c98eb2
5 changed files with 526 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelog.d/20204.bugfix
+1
View File
@@ -0,0 +1 @@
Add support for un-soft-failing [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) when room state changes, making federation support more reliable.
synapse/storage/databases/main/events.py
+30
View File
@@ -23,6 +23,7 @@ import collections
import itertools
import logging
from collections import OrderedDict
from collections.abc import Set
from typing import (
TYPE_CHECKING,
Any,
@@ -379,6 +380,24 @@ class PersistEventsStore:
)
)
sticky_events_to_un_soft_fail: set[str] = set()
if self._msc4354_enabled and state_delta_for_room is not None:
# When we change the room's current state with `state_delta_for_room`,
# that might cause some previously soft-failed sticky events to now pass
# the state-dependent auth checks.
# In other words, the sticky events could have been valid if they had
# waited for these state changes.
# For that reason, we give sticky events a second chance.
# We compute them here and then un-soft-fail them atomically with the
# persistence of the events.
sticky_events_to_un_soft_fail = (
await self.store.compute_sticky_events_to_un_soft_fail(
room_id,
events_and_contexts,
state_delta_for_room,
)
)
await self.db_pool.runInteraction(
"persist_events",
self._persist_events_txn,
@@ -390,6 +409,7 @@ class PersistEventsStore:
new_event_links=new_event_links,
sliding_sync_table_changes=sliding_sync_table_changes,
new_state_dag_forward_extremities=new_state_dag_forward_extremities,
sticky_events_to_un_soft_fail=sticky_events_to_un_soft_fail,
)
persist_event_counter.labels(**{SERVER_NAME_LABEL: self.server_name}).inc(
len(events_and_contexts)
@@ -1055,6 +1075,7 @@ class PersistEventsStore:
new_event_links: dict[str, NewEventChainLinks],
sliding_sync_table_changes: SlidingSyncTableChanges | None,
new_state_dag_forward_extremities: set[str] | None = None,
sticky_events_to_un_soft_fail: Set[str] = frozenset(),
) -> None:
"""Insert some number of room events into the necessary database tables.
@@ -1083,6 +1104,8 @@ class PersistEventsStore:
`sliding_sync_membership_snapshots` and `sliding_sync_joined_rooms` tables
derived from the given `delta_state` (see
`_calculate_sliding_sync_table_changes(...)`)
sticky_events_to_un_soft_fail:
Sticky events which will be un-soft-failed when persisting the events.
Raises:
PartialStateConflictError: if attempting to persist a partial state event in
@@ -1213,6 +1236,13 @@ class PersistEventsStore:
txn, [ev for ev, _ in events_and_contexts]
)
# Un-soft-fail any sticky events that the state delta applied just above
# has made valid.
if sticky_events_to_un_soft_fail:
self.store.un_soft_fail_sticky_events_txn(
txn, sticky_events_to_un_soft_fail
)
# We only update the sliding sync tables for non-backfilled events.
self._update_sliding_sync_tables_with_new_persisted_events_txn(
txn, room_id, events_and_contexts
synapse/storage/databases/main/events_worker.py
+15 -1
View File
@@ -68,7 +68,12 @@ from synapse.metrics import SERVER_NAME_LABEL
from synapse.metrics.background_process_metrics import (
wrap_as_background_process,
)
from synapse.replication.tcp.streams import BackfillStream, UnPartialStatedEventStream
from synapse.replication.tcp.streams import (
BackfillStream,
StickyEventsStream,
UnPartialStatedEventStream,
)
from synapse.replication.tcp.streams._base import 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
@@ -470,6 +475,15 @@ 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)
# A sticky event only gets a new row on this stream when it is first
# persisted (in which case there's nothing cached to invalidate) or when
# its soft-failure status changed, which is stored in the event's
# internal metadata, so invalidate the cached event.
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
+340 -1
View File
@@ -12,12 +12,18 @@
# <https://www.gnu.org/licenses/agpl-3.0.html>.
import logging
import random
from collections.abc import Set
from dataclasses import dataclass
from itertools import chain
from typing import TYPE_CHECKING, Collection, cast
from twisted.internet.defer import Deferred
from synapse import event_auth
from synapse.api.constants import EventTypes
from synapse.api.errors import AuthError
from synapse.events import EventBase
from synapse.events.snapshot import EventPersistencePair
from synapse.replication.tcp.streams._base import StickyEventsStream
from synapse.storage.database import (
DatabasePool,
@@ -26,10 +32,14 @@ from synapse.storage.database import (
make_in_list_sql_clause,
)
from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore
from synapse.storage.databases.main.events import DeltaState
from synapse.storage.databases.main.state import StateGroupWorkerStore
from synapse.storage.engines import PostgresEngine, Sqlite3Engine
from synapse.storage.util.id_generators import MultiWriterIdGenerator
from synapse.types import StateKey
from synapse.types.state import StateFilter
from synapse.util.duration import Duration
from synapse.util.stringutils import shortstr
if TYPE_CHECKING:
from synapse.server import HomeServer
@@ -127,7 +137,7 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
super().process_replication_position(stream_name, instance_name, token)
def get_max_sticky_events_stream_id(self) -> int:
"""Get the current maximum stream_id for thread subscriptions.
"""Get the current maximum stream_id for sticky events.
Returns:
The maximum stream_id
@@ -391,6 +401,335 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
],
)
async def compute_sticky_events_to_un_soft_fail(
self,
room_id: str,
events_and_contexts: list[EventPersistencePair],
state_delta_for_room: DeltaState,
) -> set[str]:
"""
Determine which soft-failed sticky events in the given room will become
un-soft-failed once `state_delta_for_room` has been applied to the current state.
As per MSC4354:
> **Re-evaluate soft-failure** of soft-failed unexpired sticky events when the membership state of the sender changes.[^softfail]
>
> [^softfail]: Not all servers will agree on soft-failure status due to the check considering the “current state” of the room.
> To ensure all servers agree on which events are sticky, we need to re-evaluate soft-failed status when the current room state changes.
> This becomes particularly important when room state is rolled back. For example, if Charlie sends some sticky event E and
> then Bob kicks Charlie, but concurrently Alice kicks Bob then whether or not a receiving server would accept E would depend
> on whether they saw “Alice kicks Bob” or “Bob kicks Charlie”. If they saw “Alice kicks Bob” then E would be accepted. If they
> saw “Bob kicks Charlie” then E would be rejected, and would need to be rolled back when they see “Alice kicks Bob”.
>
> — https://github.com/matrix-org/matrix-spec-proposals/blob/4ad14b0cd3b09205dcba59e45cbf1cab1e75edf7/proposals/4354-sticky-events.md#L95
Must be called from within the per-room event persistence critical section (see
`_EventPeristenceQueue`) and immediately before the persist transaction, so that
nothing else can change the room's current state in the meantime.
Args:
room_id: The room that all of the events belong to
events_and_contexts: The events about to be persisted. These are not eligible
for re-evaluation.
state_delta_for_room: The changes about to be made to the current state, used
to detect if we need to re-evaluate soft-failed sticky events.
Returns:
The event IDs of sticky events which are currently recorded as soft-failed
but which pass auth against the new current state.
"""
assert self._can_write_to_sticky_events
# Fetch the soft-failed sticky events to recheck
event_ids_to_check = await self._get_soft_failed_sticky_events_to_recheck(
room_id, state_delta_for_room
)
# Defensively filter out soft-failed events in events_and_contexts: they haven't been
# inserted into `sticky_events` yet, but be defensive in case we are asked to
# re-persist an event which is already there (e.g. de-outliering), as their
# soft failure status won't have changed for them.
persisting_event_ids = {ev.event_id for ev, _ in events_and_contexts}
event_ids_to_check = [
event_id
for event_id in event_ids_to_check
if event_id not in persisting_event_ids
]
if not event_ids_to_check:
return set()
events_to_check = await self.get_events(
event_ids_to_check, allow_rejected=False
)
# Calculate what (state event type, state key) tuples are needed as auth events for the
# soft-failed events we are reconsidering?
# e.g. [('m.room.member', '@user:example.org'), ('m.room.power_levels', ''), ...]
needed_state_tuples_for_auth: set[StateKey] = set()
for soft_failed_event in events_to_check.values():
needed_state_tuples_for_auth.update(
event_auth.auth_types_for_event(
soft_failed_event.room_version, soft_failed_event
)
)
# Load the needed auth state from the current state
# (type, state key) -> event_id
current_auth_state_ids_map = dict(
await self.get_partial_filtered_current_state_ids(
room_id, StateFilter.from_types(needed_state_tuples_for_auth)
)
)
# `state_delta_for_room` hasn't yet been applied to the room's persisted current state,
# so we need to apply it here to the auth state we are using for the re-evaluation
for deleted_key in state_delta_for_room.to_delete:
current_auth_state_ids_map.pop(deleted_key, None)
for inserted_key, inserted_event_id in state_delta_for_room.to_insert.items():
if inserted_key in needed_state_tuples_for_auth:
current_auth_state_ids_map[inserted_key] = inserted_event_id
# Now load in the auth events
persisting_events_by_id = {ev.event_id: ev for ev, _ in events_and_contexts}
current_auth_events: list[EventBase] = []
current_auth_state_event_ids_to_fetch: list[str] = []
for event_id in current_auth_state_ids_map.values():
persisting_event = persisting_events_by_id.get(event_id)
if persisting_event is not None:
# This event is one we are about to persist, so just use it
current_auth_events.append(persisting_event)
else:
# This event needs to be loaded from the database
current_auth_state_event_ids_to_fetch.append(event_id)
current_auth_events.extend(
await self.get_events_as_list(current_auth_state_event_ids_to_fetch)
)
passing_event_ids: set[str] = set()
for soft_failed_event in events_to_check.values():
try:
# We don't need to check_state_independent_auth_rules as that doesn't depend on room state,
# so if it passed once it'll pass again.
event_auth.check_state_dependent_auth_rules(
soft_failed_event, current_auth_events
)
# Ready to be un-soft-failed
passing_event_ids.add(soft_failed_event.event_id)
except AuthError:
# state-dependent auth rules still unsatisfied: remain soft-failed
pass
if passing_event_ids:
logger.info(
"%s soft-failed events now pass current state checks in room %s : %s",
len(passing_event_ids),
room_id,
shortstr(passing_event_ids),
)
return passing_event_ids
async def _get_soft_failed_sticky_events_to_recheck(
self,
room_id: str,
state_delta_for_room: DeltaState,
) -> list[str]:
"""
Fetch soft-failed sticky events which should be rechecked against the current state.
Returns:
A list of event IDs to recheck
"""
if state_delta_for_room.no_longer_in_room:
# We're leaving the room, so the current state is about to be wiped and
# nothing can pass auth against it.
return []
# Only a change to critical auth state may change soft failure status.
# This means any changes to join rules, power levels or member events.
# If the state has changed but these types are unchanged, we don't need to recheck.
CRITICAL_AUTH_TYPES = (
EventTypes.JoinRules,
EventTypes.PowerLevels,
EventTypes.Member,
)
critical_auth_types_changed = {
typ
for typ, _ in chain(
state_delta_for_room.to_insert, state_delta_for_room.to_delete
)
if typ in CRITICAL_AUTH_TYPES
}
if len(critical_auth_types_changed) == 0:
# No change to critical auth events.
# No way soft failure status could be different.
return []
if critical_auth_types_changed == {EventTypes.Member}:
# If the only critical auth state that changed is user memberships,
# then we can restrict our re-evaluation to only reconsider soft-failed sticky events sent
# by the users who have their membership changed.
# Events sent by any other user can not be affected,
# with the pedantic yet possible exception of sticky invite/kick/ban `m.room.member`
# state events (where state key ≠ sender).
# That said: we don't expect to use those and it is not possible to create one with
# the Client-Server API.
changed_members = {
membership_user_id
for event_type, membership_user_id in chain(
state_delta_for_room.to_insert, state_delta_for_room.to_delete
)
if event_type == EventTypes.Member
}
return await self.db_pool.runInteraction(
"_get_soft_failed_sticky_events_to_recheck_members",
self._get_soft_failed_sticky_events_txn,
room_id,
# Only reconsider events from changed members
senders=changed_members,
)
# If we reach here, then it must be the case that there have been changes in
# power level or join rules.
# In both of these cases we want to re-evaluate soft failure status of all the
# soft-failed events in the room.
#
# NB: event auth checks are NOT recursive. We don't need to specifically handle the case where
# an admin user's membership changes which causes a PL event to be allowed, as when the PL event
# gets allowed we will re-evaluate anyway. E.g:
#
# PL(send_event=0, sender=Admin) #1
# ^ ^_____________________
# | |
# . PL(send_event=50, sender=Mod) #2 sticky event (sender=User) #3
#
# In this scenario, the sticky event is soft-failed due to the Mod updating the PL event to
# set send_event=50, which User does not have. If we learn of an event which makes Mod's PL
# event invalid (say, Mod was banned by Admin concurrently to Mod setting the PL event), then
# the act of seeing the ban event will cause the old PL event to be in the state delta, meaning
# we will re-evaluate the sticky event due to the PL changing. We don't need to specially handle
# this case.
return await self.db_pool.runInteraction(
"_get_soft_failed_sticky_events_to_recheck",
self._get_soft_failed_sticky_events_txn,
room_id,
# Consider everyone
senders=None,
)
def _get_soft_failed_sticky_events_txn(
self,
txn: LoggingTransaction,
room_id: str,
*,
senders: Collection[str] | None,
) -> list[str]:
"""
Fetch the event IDs of (unexpired) soft-failed sticky events in a room.
Args:
room_id: the room to look in.
senders:
If present, only return sticky events sent by one of these users.
If None, do not restrict by sender.
"""
sender_clause = ""
sender_args: Collection[str] = ()
if senders is not None:
if not senders:
return []
sender_sql, sender_args = make_in_list_sql_clause(
txn.database_engine, "se.sender", senders
)
sender_clause = f"AND {sender_sql}"
if isinstance(self.database_engine, PostgresEngine):
expr_soft_failed = "COALESCE(((ej.internal_metadata::jsonb)->>'soft_failed')::boolean, FALSE)"
else:
expr_soft_failed = "COALESCE(ej.internal_metadata->>'soft_failed', FALSE)"
# Note that we are relying on the 1h stickiness limit to make this
# tractable, as we can't realistically apply any LIMIT here.
txn.execute(
f"""
SELECT se.event_id
FROM sticky_events se
INNER JOIN event_json ej USING (event_id)
WHERE
se.room_id = ?
AND ? < se.expires_at
AND {expr_soft_failed}
{sender_clause}
""",
(room_id, self.clock.time_msec(), *sender_args),
)
return [event_id for (event_id,) in txn]
def un_soft_fail_sticky_events_txn(
self, txn: LoggingTransaction, sticky_event_ids: Set[str]
) -> None:
"""
For the given soft-failed sticky events:
- removes their soft-failed status
- moves them to the end of the `sticky_events` stream so that clients get told about them
"""
if not sticky_event_ids:
return
# Update the internal metadata on the event itself.
event_id_in_list_clause, event_id_in_list_args = make_in_list_sql_clause(
txn.database_engine,
"event_id",
sticky_event_ids,
)
if isinstance(txn.database_engine, PostgresEngine):
# It's a bit sad that internal_metadata is TEXT and not JSONB...
txn.execute(
f"""
UPDATE event_json
SET internal_metadata = (
jsonb_set(internal_metadata::jsonb, '{{soft_failed}}', 'false'::jsonb)
)::text
WHERE {event_id_in_list_clause}
""",
event_id_in_list_args,
)
else:
assert isinstance(txn.database_engine, Sqlite3Engine)
txn.execute(
f"""
UPDATE event_json
SET internal_metadata = json_set(internal_metadata, '$.soft_failed', json('false'))
WHERE {event_id_in_list_clause}
""",
event_id_in_list_args,
)
# Invalidate caches as a result
for event_id in sticky_event_ids:
self.invalidate_get_event_cache_after_txn(txn, event_id)
# Move the events to the end of the sticky events stream
new_stream_ids = self._sticky_events_id_gen.get_next_mult_txn(
txn, len(sticky_event_ids)
)
self.db_pool.simple_update_many_txn(
txn,
table="sticky_events",
key_names=("event_id",),
key_values=[(event_id,) for event_id in sticky_event_ids],
value_names=(
"stream_id",
"instance_name",
),
value_values=[
(stream_id, self._instance_name) for stream_id in new_stream_ids
],
)
async def _delete_expired_sticky_events(self) -> None:
await self.db_pool.runInteraction(
"_delete_expired_sticky_events",
tests/storage/test_sticky_events.py
+140 -2
View File
@@ -11,6 +11,7 @@
# See the GNU Affero General Public License for more details:
# <https://www.gnu.org/licenses/agpl-3.0.html>.
import sqlite3
from http import HTTPStatus
from twisted.internet.testing import MemoryReactor
@@ -23,7 +24,7 @@ from synapse.api.constants import (
)
from synapse.api.room_versions import RoomVersions
from synapse.rest import admin
from synapse.rest.client import login, register, room
from synapse.rest.client import login, register, room, sync
from synapse.server import HomeServer
from synapse.types import JsonDict, create_requester
from synapse.util.clock import Clock
@@ -45,6 +46,7 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
servlets = [
room.register_servlets,
sync.register_servlets,
login.register_servlets,
register.register_servlets,
admin.register_servlets,
@@ -52,7 +54,10 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
def default_config(self) -> JsonDict:
config = super().default_config()
config["experimental_features"] = {"msc4354_enabled": True}
config["experimental_features"] = {
"msc3575_enabled": True,
"msc4354_enabled": True,
}
return config
def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
@@ -428,3 +433,136 @@ class StickyEventsTestCase(unittest.HomeserverTestCase):
self.assertEqual(len(updates), 1)
self.assertEqual(updates[0].event_id, valid_sticky_event.event_id)
def _get_visible_sticky_event_ids(self) -> set[str]:
"""
Returns the IDs of the sticky events visible to clients in sync.
"""
sync_body: JsonDict = {
"lists": {
"main": {
"ranges": [[0, 0]],
"required_state": [],
# We don't want any timeline events, just sticky events
"timeline_limit": 0,
}
},
"extensions": {
"org.matrix.msc4354.sticky_events": {
"enabled": True,
}
},
}
channel = self.make_request(
"POST",
"/_matrix/client/unstable/org.matrix.simplified_msc3575/sync",
sync_body,
access_token=self.token,
)
self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
sticky_events = channel.json_body["extensions"].get(
"org.matrix.msc4354.sticky_events"
)
if sticky_events is None:
return set()
events_in_room = (
sticky_events.get("rooms", {}).get(self.room_id, {}).get("events", [])
)
return {event["event_id"] for event in events_in_room}
def test_soft_failure_cleared_when_state_changes(self) -> None:
"""
Tests that a soft-failed sticky event stops being soft-failed once a change to
the room's current state means that it passes auth after all.
"""
user2_id = self.register_user("user2", "pass")
user2_tok = self.login(user2_id, "pass")
self.helper.join(self.room_id, user2_id, tok=user2_tok)
# Devoice user2, so that their sticky event will (realistically) fail auth
# against the room's current state.
self.helper.send_state(
self.room_id,
EventTypes.PowerLevels,
body={"users": {self.user_id: 100, user2_id: -1}, "events_default": 0},
tok=self.token,
)
# Inject a soft-failed sticky event. This is cheating a bit for brevity.
# In the real world, we'd need to craft a soft-failed sticky event to arrive over federation.
# The Complement test will do this: https://github.com/matrix-org/complement/pull/806/files#diff-6c9d6d169485d0848c6b20dd9b43f6fe669a8a710e42f953d08fa25a99cc8f4cR509
event_id = self.get_success(
inject_event(
self.hs,
room_id=self.room_id,
sender=user2_id,
type=EventTypes.Message,
content={"body": "sticky", "msgtype": "m.text"},
internal_metadata={"soft_failed": True},
# Corresponds to StickyEvent.EVENT_FIELD_NAME
msc4354_sticky=StickyEventField(
duration_ms=Duration(minutes=1).as_millis()
),
)
).event_id
# Whilst it is soft-failed, the event isn't shown to clients.
self.assertEqual(self._get_visible_sticky_event_ids(), set())
# Change the room's power levels to voice user2 back.
# This triggers the soft-fail re-evaluation and also allows the soft-failed sticky
# event to pass state-dependent auth checks against the current state, becoming
# un-soft-failed
self.helper.send_state(
self.room_id,
EventTypes.PowerLevels,
body={"users": {self.user_id: 100, user2_id: 0}, "events_default": 0},
tok=self.token,
)
# The event has been re-evaluated and is now shown to clients...
self.assertEqual(self._get_visible_sticky_event_ids(), {event_id})
# ...and the soft-failure flag has been cleared.
event = self.get_success(self.store.get_event(event_id))
self.assertFalse(event.internal_metadata.is_soft_failed())
def test_soft_failure_cleared_when_sender_membership_changes(self) -> None:
"""
Tests that soft-failure status of a sticky event is reconsidered when
the sender's membership changes.
"""
user2_id = self.register_user("user2", "pass")
user2_tok = self.login(user2_id, "pass")
# Inject a soft-failed sticky event from user2
event_id = self.get_success(
inject_event(
self.hs,
room_id=self.room_id,
sender=user2_id,
type=EventTypes.Message,
content={"body": "sticky", "msgtype": "m.text"},
internal_metadata={"soft_failed": True},
# Corresponds to StickyEvent.EVENT_FIELD_NAME
msc4354_sticky=StickyEventField(
duration_ms=Duration(minutes=1).as_millis()
),
)
).event_id
# Whilst it is soft-failed, the event isn't shown to clients.
self.assertEqual(self._get_visible_sticky_event_ids(), set())
# Check that an irrelevant user's membership changing doesn't affect the event
user3_id = self.register_user("user3", "pass")
user3_tok = self.login(user3_id, "pass")
self.helper.join(self.room_id, user3_id, tok=user3_tok)
self.assertEqual(self._get_visible_sticky_event_ids(), set())
# The sender joins, so the event now passes auth and is un-soft-failed.
self.helper.join(self.room_id, user2_id, tok=user2_tok)
self.assertEqual(self._get_visible_sticky_event_ids(), {event_id})
# ...and the soft-failure has been cleared from the event itself.
event = self.get_success(self.store.get_event(event_id))
self.assertFalse(event.internal_metadata.is_soft_failed())