mirror of
https://github.com/element-hq/synapse.git
synced 2026-08-29 05:38:26 +00:00
Send sticky events to newly joined servers.
Process those sticky events as well. In so doing, fixes https://github.com/element-hq/synapse/issues/18563 because we now have the room_creator as an update value not insert value.
This commit is contained in:
Kegan Dougal
@@ -181,6 +181,7 @@ from synapse.types import (
|
||||
from synapse.util.clock import Clock
|
||||
from synapse.util.metrics import Measure
|
||||
from synapse.util.retryutils import filter_destinations_by_retry_limiter
|
||||
from synapse.visibility import filter_events_for_server
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from synapse.events.presence_router import PresenceRouter
|
||||
@@ -496,7 +497,58 @@ class FederationSender(AbstractFederationSender):
|
||||
return queue
|
||||
|
||||
def notify_new_server_joined(self, server: str, room_id: str) -> None:
|
||||
print(f"FEDSENDER: new server joined: server={server} room={room_id}")
|
||||
# We currently only use this notification for MSC4354: Sticky Events.
|
||||
if not self.hs.config.experimental.msc4354_enabled:
|
||||
return
|
||||
# fire off a processing loop in the background
|
||||
run_as_background_process(
|
||||
"process_new_server_joined_over_federation",
|
||||
self.server_name,
|
||||
self._process_new_server_joined_over_federation,
|
||||
server,
|
||||
room_id,
|
||||
)
|
||||
|
||||
async def _process_new_server_joined_over_federation(
|
||||
self, new_server: str, room_id: str
|
||||
) -> None:
|
||||
sticky_event_ids = await self.store.get_sticky_event_ids_sent_by_self(
|
||||
room_id,
|
||||
0,
|
||||
)
|
||||
sticky_events = await self.store.get_events_as_list(sticky_event_ids)
|
||||
|
||||
# We must not send events that are outliers / lack a stream ordering, else we won't be able to
|
||||
# satisfy /get_missing_events requests
|
||||
sticky_events = [
|
||||
ev
|
||||
for ev in sticky_events
|
||||
if ev.internal_metadata.stream_ordering is not None
|
||||
and not ev.internal_metadata.is_outlier()
|
||||
]
|
||||
# order by stream ordering so we present things in the right timeline order on the receiver
|
||||
sticky_events = sorted(
|
||||
sticky_events,
|
||||
key=lambda ev: ev.internal_metadata.stream_ordering
|
||||
or 0, # not possible to be 0
|
||||
)
|
||||
|
||||
sticky_events = await filter_events_for_server(
|
||||
self._storage_controllers,
|
||||
new_server,
|
||||
self.server_name,
|
||||
sticky_events,
|
||||
redact=False,
|
||||
filter_out_erased_senders=True,
|
||||
filter_out_remote_partial_state_events=True,
|
||||
)
|
||||
if sticky_events:
|
||||
logger.info("sending %d sticky events to newly joined server %s in room %s", len(sticky_events), new_server, room_id)
|
||||
# we don't track that we sent up to this stream position since it won't make any difference
|
||||
# since notify_new_server_joined is only called initially.
|
||||
await self._transaction_manager.send_new_transaction(
|
||||
new_server, sticky_events, []
|
||||
)
|
||||
|
||||
def notify_new_events(self, max_token: RoomStreamToken) -> None:
|
||||
"""This gets called when we have some new events we might want to
|
||||
|
||||
+148
-113
@@ -67,6 +67,7 @@ from synapse.events import EventBase
|
||||
from synapse.events.snapshot import EventContext, UnpersistedEventContextBase
|
||||
from synapse.events.validator import EventValidator
|
||||
from synapse.federation.federation_client import InvalidResponseError
|
||||
from synapse.federation.federation_server import _INBOUND_EVENT_HANDLING_LOCK_NAME
|
||||
from synapse.handlers.pagination import PURGE_PAGINATION_LOCK_NAME
|
||||
from synapse.http.servlet import assert_params_in_dict
|
||||
from synapse.logging.context import nested_logging_context
|
||||
@@ -75,6 +76,7 @@ from synapse.metrics import SERVER_NAME_LABEL
|
||||
from synapse.metrics.background_process_metrics import run_as_background_process
|
||||
from synapse.module_api import NOT_SPAM
|
||||
from synapse.storage.databases.main.events_worker import EventRedactBehaviour
|
||||
from synapse.storage.databases.main.lock import Lock
|
||||
from synapse.storage.invite_rule import InviteRule
|
||||
from synapse.types import JsonDict, StrCollection, get_domain_from_id
|
||||
from synapse.types.state import StateFilter
|
||||
@@ -647,125 +649,158 @@ class FederationHandler:
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
lock: Optional[Lock] = None
|
||||
async with self._is_partial_state_room_linearizer.queue(room_id):
|
||||
already_partial_state_room = await self.store.is_partial_state_room(
|
||||
room_id
|
||||
)
|
||||
|
||||
ret = await self.federation_client.send_join(
|
||||
host_list,
|
||||
event,
|
||||
room_version_obj,
|
||||
# Perform a full join when we are already in the room and it is a
|
||||
# full state room, since we are not allowed to persist a partial
|
||||
# state join event in a full state room. In the future, we could
|
||||
# optimize this by always performing a partial state join and
|
||||
# computing the state ourselves or retrieving it from the remote
|
||||
# homeserver if necessary.
|
||||
#
|
||||
# There's a race where we leave the room, then perform a full join
|
||||
# anyway. This should end up being fast anyway, since we would
|
||||
# already have the full room state and auth chain persisted.
|
||||
partial_state=not is_host_joined or already_partial_state_room,
|
||||
)
|
||||
|
||||
event = ret.event
|
||||
origin = ret.origin
|
||||
state = ret.state
|
||||
auth_chain = ret.auth_chain
|
||||
auth_chain.sort(key=lambda e: e.depth)
|
||||
|
||||
logger.debug("do_invite_join auth_chain: %s", auth_chain)
|
||||
logger.debug("do_invite_join state: %s", state)
|
||||
|
||||
logger.debug("do_invite_join event: %s", event)
|
||||
|
||||
# if this is the first time we've joined this room, it's time to add
|
||||
# a row to `rooms` with the correct room version. If there's already a
|
||||
# row there, we should override it, since it may have been populated
|
||||
# based on an invite request which lied about the room version.
|
||||
#
|
||||
# federation_client.send_join has already checked that the room
|
||||
# version in the received create event is the same as room_version_obj,
|
||||
# so we can rely on it now.
|
||||
#
|
||||
await self.store.upsert_room_on_join(
|
||||
room_id=room_id,
|
||||
room_version=room_version_obj,
|
||||
state_events=state,
|
||||
)
|
||||
|
||||
if ret.partial_state and not already_partial_state_room:
|
||||
# Mark the room as having partial state.
|
||||
# The background process is responsible for unmarking this flag,
|
||||
# even if the join fails.
|
||||
# TODO(faster_joins):
|
||||
# We may want to reset the partial state info if it's from an
|
||||
# old, failed partial state join.
|
||||
# https://github.com/matrix-org/synapse/issues/13000
|
||||
|
||||
# FIXME: Ideally, we would store the full stream token here
|
||||
# not just the minimum stream ID, so that we can compute an
|
||||
# accurate list of device changes when un-partial-ing the
|
||||
# room. The only side effect of this is that we may send
|
||||
# extra unecessary device list outbound pokes through
|
||||
# federation, which is harmless.
|
||||
device_lists_stream_id = self.store.get_device_stream_token().stream
|
||||
|
||||
await self.store.store_partial_state_room(
|
||||
room_id=room_id,
|
||||
servers=ret.servers_in_room,
|
||||
device_lists_stream_id=device_lists_stream_id,
|
||||
joined_via=origin,
|
||||
)
|
||||
|
||||
try:
|
||||
max_stream_id = (
|
||||
await self._federation_event_handler.process_remote_join(
|
||||
origin,
|
||||
room_id,
|
||||
auth_chain,
|
||||
state,
|
||||
event,
|
||||
room_version_obj,
|
||||
partial_state=ret.partial_state,
|
||||
# MSC4354: Sticky Events causes existing servers in the room to send sticky events
|
||||
# to the newly joined server as soon as they realise the new server is in the room.
|
||||
# If they do this before we've persisted the /send_join response we will be unable to
|
||||
# process those PDUs. Therefore, we take a lock out now for this room, and release it
|
||||
# once we have processed the /send_join response, to buffer up these inbound messages.
|
||||
# This may be useful to do even without MSC4354, but it's gated behind an
|
||||
# experimental flag check to reduce the chance of this having unintended side-effects
|
||||
# e.g accidental deadlocks. Once we're confident of this behaviour, we can probably
|
||||
# drop the flag check. We take the lock AFTER we have been queued by the linearizer
|
||||
# else we would just hold the lock for no reason whilst in the queue: we want to hold
|
||||
# the lock for the smallest amount of time possible.
|
||||
if self.config.experimental.msc4354_enabled:
|
||||
lock = await self.store.try_acquire_lock(
|
||||
_INBOUND_EVENT_HANDLING_LOCK_NAME, room_id
|
||||
)
|
||||
)
|
||||
except PartialStateConflictError:
|
||||
# This should be impossible, since we hold the lock on the room's
|
||||
# partial statedness.
|
||||
logger.error(
|
||||
"Room %s was un-partial stated while processing remote join.",
|
||||
room_id,
|
||||
)
|
||||
raise
|
||||
else:
|
||||
# Record the join event id for future use (when we finish the full
|
||||
# join). We have to do this after persisting the event to keep
|
||||
# foreign key constraints intact.
|
||||
if ret.partial_state and not already_partial_state_room:
|
||||
# TODO(faster_joins):
|
||||
# We may want to reset the partial state info if it's from
|
||||
# an old, failed partial state join.
|
||||
# https://github.com/matrix-org/synapse/issues/13000
|
||||
await self.store.write_partial_state_rooms_join_event_id(
|
||||
room_id, event.event_id
|
||||
)
|
||||
finally:
|
||||
# Always kick off the background process that asynchronously fetches
|
||||
# state for the room.
|
||||
# If the join failed, the background process is responsible for
|
||||
# cleaning up — including unmarking the room as a partial state
|
||||
# room.
|
||||
if ret.partial_state:
|
||||
# Kick off the process of asynchronously fetching the state for
|
||||
# this room.
|
||||
self._start_partial_state_room_sync(
|
||||
initial_destination=origin,
|
||||
other_destinations=ret.servers_in_room,
|
||||
# Insert the room into the rooms table now so we can process potential incoming
|
||||
# /send transactions enough to be able to insert into the federation staging
|
||||
# area. We won't process the staging area until we release the lock above.
|
||||
await self.store.upsert_room_on_join(
|
||||
room_id=room_id,
|
||||
room_version=room_version_obj,
|
||||
state_events=None,
|
||||
)
|
||||
|
||||
already_partial_state_room = await self.store.is_partial_state_room(
|
||||
room_id
|
||||
)
|
||||
|
||||
ret = await self.federation_client.send_join(
|
||||
host_list,
|
||||
event,
|
||||
room_version_obj,
|
||||
# Perform a full join when we are already in the room and it is a
|
||||
# full state room, since we are not allowed to persist a partial
|
||||
# state join event in a full state room. In the future, we could
|
||||
# optimize this by always performing a partial state join and
|
||||
# computing the state ourselves or retrieving it from the remote
|
||||
# homeserver if necessary.
|
||||
#
|
||||
# There's a race where we leave the room, then perform a full join
|
||||
# anyway. This should end up being fast anyway, since we would
|
||||
# already have the full room state and auth chain persisted.
|
||||
partial_state=not is_host_joined or already_partial_state_room,
|
||||
)
|
||||
|
||||
event = ret.event
|
||||
origin = ret.origin
|
||||
state = ret.state
|
||||
auth_chain = ret.auth_chain
|
||||
auth_chain.sort(key=lambda e: e.depth)
|
||||
|
||||
logger.debug("do_invite_join auth_chain: %s", auth_chain)
|
||||
logger.debug("do_invite_join state: %s", state)
|
||||
|
||||
logger.debug("do_invite_join event: %s", event)
|
||||
|
||||
# if this is the first time we've joined this room, it's time to add
|
||||
# a row to `rooms` with the correct room version. If there's already a
|
||||
# row there, we should override it, since it may have been populated
|
||||
# based on an invite request which lied about the room version.
|
||||
#
|
||||
# federation_client.send_join has already checked that the room
|
||||
# version in the received create event is the same as room_version_obj,
|
||||
# so we can rely on it now.
|
||||
#
|
||||
await self.store.upsert_room_on_join(
|
||||
room_id=room_id,
|
||||
room_version=room_version_obj,
|
||||
state_events=state,
|
||||
)
|
||||
|
||||
if ret.partial_state and not already_partial_state_room:
|
||||
# Mark the room as having partial state.
|
||||
# The background process is responsible for unmarking this flag,
|
||||
# even if the join fails.
|
||||
# TODO(faster_joins):
|
||||
# We may want to reset the partial state info if it's from an
|
||||
# old, failed partial state join.
|
||||
# https://github.com/matrix-org/synapse/issues/13000
|
||||
|
||||
# FIXME: Ideally, we would store the full stream token here
|
||||
# not just the minimum stream ID, so that we can compute an
|
||||
# accurate list of device changes when un-partial-ing the
|
||||
# room. The only side effect of this is that we may send
|
||||
# extra unecessary device list outbound pokes through
|
||||
# federation, which is harmless.
|
||||
device_lists_stream_id = (
|
||||
self.store.get_device_stream_token().stream
|
||||
)
|
||||
|
||||
await self.store.store_partial_state_room(
|
||||
room_id=room_id,
|
||||
servers=ret.servers_in_room,
|
||||
device_lists_stream_id=device_lists_stream_id,
|
||||
joined_via=origin,
|
||||
)
|
||||
|
||||
try:
|
||||
max_stream_id = (
|
||||
await self._federation_event_handler.process_remote_join(
|
||||
origin,
|
||||
room_id,
|
||||
auth_chain,
|
||||
state,
|
||||
event,
|
||||
room_version_obj,
|
||||
partial_state=ret.partial_state,
|
||||
)
|
||||
)
|
||||
except PartialStateConflictError:
|
||||
# This should be impossible, since we hold the lock on the room's
|
||||
# partial statedness.
|
||||
logger.error(
|
||||
"Room %s was un-partial stated while processing remote join.",
|
||||
room_id,
|
||||
)
|
||||
raise
|
||||
else:
|
||||
# Record the join event id for future use (when we finish the full
|
||||
# join). We have to do this after persisting the event to keep
|
||||
# foreign key constraints intact.
|
||||
if ret.partial_state and not already_partial_state_room:
|
||||
# TODO(faster_joins):
|
||||
# We may want to reset the partial state info if it's from
|
||||
# an old, failed partial state join.
|
||||
# https://github.com/matrix-org/synapse/issues/13000
|
||||
await self.store.write_partial_state_rooms_join_event_id(
|
||||
room_id, event.event_id
|
||||
)
|
||||
finally:
|
||||
# Always kick off the background process that asynchronously fetches
|
||||
# state for the room.
|
||||
# If the join failed, the background process is responsible for
|
||||
# cleaning up — including unmarking the room as a partial state
|
||||
# room.
|
||||
if ret.partial_state:
|
||||
# Kick off the process of asynchronously fetching the state for
|
||||
# this room.
|
||||
self._start_partial_state_room_sync(
|
||||
initial_destination=origin,
|
||||
other_destinations=ret.servers_in_room,
|
||||
room_id=room_id,
|
||||
)
|
||||
finally:
|
||||
# allow inbound events which happened during the join to be processed.
|
||||
# Also ensures we release the lock on unexpected errors e.g db errors from
|
||||
# upsert_room_on_join or network errors from send_join.
|
||||
if lock:
|
||||
await lock.release()
|
||||
# We wait here until this instance has seen the events come down
|
||||
# replication (if we're using replication) as the below uses caches.
|
||||
await self._replication.wait_for_stream_position(
|
||||
|
||||
@@ -1187,7 +1187,9 @@ class PersistEventsStore:
|
||||
)
|
||||
|
||||
if self.msc4354_sticky_events:
|
||||
self.store.insert_sticky_events_txn(txn, events_and_contexts)
|
||||
self.store.insert_sticky_events_txn(
|
||||
txn, [ev for ev, _ in events_and_contexts]
|
||||
)
|
||||
for ev, _ in events_and_contexts:
|
||||
if ev.type == "m.room.member" and ev.membership == "join":
|
||||
print(f"GOT JOIN FOR {ev.state_key}")
|
||||
@@ -2658,6 +2660,11 @@ class PersistEventsStore:
|
||||
# event isn't an outlier any more.
|
||||
self._update_backward_extremeties(txn, [event])
|
||||
|
||||
if self.msc4354_sticky_events and event.sticky_duration():
|
||||
# The de-outliered event is sticky. Update the sticky events table to ensure
|
||||
# we delivery this down /sync.
|
||||
self.store.insert_sticky_events_txn(txn, [event])
|
||||
|
||||
return [ec for ec in events_and_contexts if ec[0] not in to_remove]
|
||||
|
||||
def _store_event_txn(
|
||||
|
||||
@@ -2460,7 +2460,10 @@ class RoomStore(RoomBackgroundUpdateStore, RoomWorkerStore):
|
||||
self._instance_name = hs.get_instance_name()
|
||||
|
||||
async def upsert_room_on_join(
|
||||
self, room_id: str, room_version: RoomVersion, state_events: List[EventBase]
|
||||
self,
|
||||
room_id: str,
|
||||
room_version: RoomVersion,
|
||||
state_events: Optional[List[EventBase]],
|
||||
) -> None:
|
||||
"""Ensure that the room is stored in the table
|
||||
|
||||
@@ -2472,36 +2475,46 @@ class RoomStore(RoomBackgroundUpdateStore, RoomWorkerStore):
|
||||
# mark the room as having an auth chain cover index.
|
||||
has_auth_chain_index = await self.has_auth_chain_index(room_id)
|
||||
|
||||
create_event = None
|
||||
for e in state_events:
|
||||
if (e.type, e.state_key) == (EventTypes.Create, ""):
|
||||
create_event = e
|
||||
break
|
||||
# We may want to insert a row into the rooms table BEFORE having the state events in the
|
||||
# room, in order to correctly handle the race condition where the /send_join is processed
|
||||
# remotely which causes remote servers to send us events before we've processed the /send_join
|
||||
# response. Therefore, we allow state_events (and thus the creator column) to be optional.
|
||||
# When we get the /send_join response, we'll patch this up.
|
||||
room_creator: Optional[str] = None
|
||||
if state_events:
|
||||
create_event = None
|
||||
for e in state_events:
|
||||
if (e.type, e.state_key) == (EventTypes.Create, ""):
|
||||
create_event = e
|
||||
break
|
||||
|
||||
if create_event is None:
|
||||
# If the state doesn't have a create event then the room is
|
||||
# invalid, and it would fail auth checks anyway.
|
||||
raise StoreError(400, "No create event in state")
|
||||
|
||||
# Before MSC2175, the room creator was a separate field.
|
||||
if not room_version.implicit_room_creator:
|
||||
room_creator = create_event.content.get(EventContentFields.ROOM_CREATOR)
|
||||
|
||||
if not isinstance(room_creator, str):
|
||||
# If the create event does not have a creator then the room is
|
||||
if create_event is None:
|
||||
# If the state doesn't have a create event then the room is
|
||||
# invalid, and it would fail auth checks anyway.
|
||||
raise StoreError(400, "No creator defined on the create event")
|
||||
else:
|
||||
room_creator = create_event.sender
|
||||
raise StoreError(400, "No create event in state")
|
||||
|
||||
# Before MSC2175, the room creator was a separate field.
|
||||
if not room_version.implicit_room_creator:
|
||||
room_creator = create_event.content.get(EventContentFields.ROOM_CREATOR)
|
||||
|
||||
if not isinstance(room_creator, str):
|
||||
# If the create event does not have a creator then the room is
|
||||
# invalid, and it would fail auth checks anyway.
|
||||
raise StoreError(400, "No creator defined on the create event")
|
||||
else:
|
||||
room_creator = create_event.sender
|
||||
|
||||
update_with = {"room_version": room_version.identifier}
|
||||
if room_creator:
|
||||
update_with["creator"] = room_creator
|
||||
|
||||
await self.db_pool.simple_upsert(
|
||||
desc="upsert_room_on_join",
|
||||
table="rooms",
|
||||
keyvalues={"room_id": room_id},
|
||||
values={"room_version": room_version.identifier},
|
||||
values=update_with,
|
||||
insertion_values={
|
||||
"is_public": False,
|
||||
"creator": room_creator,
|
||||
"has_auth_chain_index": has_auth_chain_index,
|
||||
},
|
||||
)
|
||||
|
||||
@@ -212,7 +212,7 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
|
||||
async def get_sticky_event_ids_sent_by_self(
|
||||
self, room_id: str, from_stream_pos: int
|
||||
) -> List[str]:
|
||||
"""Get sticky event IDs which have been sent by users on this homeserver.
|
||||
"""Get unexpired sticky event IDs which have been sent by users on this homeserver.
|
||||
|
||||
Used when sending sticky events eagerly to newly joined servers, or when catching up over federation.
|
||||
|
||||
@@ -284,12 +284,12 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor
|
||||
def insert_sticky_events_txn(
|
||||
self,
|
||||
txn: LoggingTransaction,
|
||||
events_and_contexts: List[EventPersistencePair],
|
||||
events: List[EventBase],
|
||||
) -> None:
|
||||
now_ms = self._now()
|
||||
# event, expires_at, stream_id
|
||||
sticky_events: List[Tuple[EventBase, int, int]] = []
|
||||
for ev, _ in events_and_contexts:
|
||||
for ev in events:
|
||||
# MSC: Note: policy servers and other similar antispam techniques still apply to these events.
|
||||
if ev.internal_metadata.policy_server_spammy:
|
||||
continue
|
||||
|
||||
Reference in New Issue
Block a user