diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py
index 21a17cd2e8..4ab4046650 100644
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -392,9 +392,6 @@ class ApplicationServicesHandler:
protocols[p].append(info)
def _merge_instances(infos: List[JsonDict]) -> JsonDict:
- if not infos:
- return {}
-
# Merge the 'instances' lists of multiple results, but just take
# the other fields from the first as they ought to be identical
# copy the result so as not to corrupt the cached one
@@ -406,7 +403,9 @@ class ApplicationServicesHandler:
return combined
- return {p: _merge_instances(protocols[p]) for p in protocols.keys()}
+ return {
+ p: _merge_instances(protocols[p]) for p in protocols.keys() if protocols[p]
+ }
async def _get_services_for_event(
self, event: EventBase
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 8197b60b76..8b602e3813 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -42,6 +42,7 @@ from twisted.internet import defer
from synapse import event_auth
from synapse.api.constants import (
+ EventContentFields,
EventTypes,
Membership,
RejectedReason,
@@ -262,7 +263,12 @@ class FederationHandler(BaseHandler):
state = None
- # Get missing pdus if necessary.
+ # Check that the event passes auth based on the state at the event. This is
+ # done for events that are to be added to the timeline (non-outliers).
+ #
+ # Get missing pdus if necessary:
+ # - Fetching any missing prev events to fill in gaps in the graph
+ # - Fetching state if we have a hole in the graph
if not pdu.internal_metadata.is_outlier():
# We only backfill backwards to the min depth.
min_depth = await self.get_min_depth_for_context(pdu.room_id)
@@ -432,6 +438,13 @@ class FederationHandler(BaseHandler):
affected=event_id,
)
+ # A second round of checks for all events. Check that the event passes auth
+ # based on `auth_events`, this allows us to assert that the event would
+ # have been allowed at some point. If an event passes this check its OK
+ # for it to be used as part of a returned `/state` request, as either
+ # a) we received the event as part of the original join and so trust it, or
+ # b) we'll do a state resolution with existing state before it becomes
+ # part of the "current state", which adds more protection.
await self._process_received_pdu(origin, pdu, state=state)
async def _get_missing_events_for_pdu(
@@ -889,6 +902,79 @@ class FederationHandler(BaseHandler):
"resync_device_due_to_pdu", self._resync_device, event.sender
)
+ await self._handle_marker_event(origin, event)
+
+ async def _handle_marker_event(self, origin: str, marker_event: EventBase):
+ """Handles backfilling the insertion event when we receive a marker
+ event that points to one.
+
+ Args:
+ origin: Origin of the event. Will be called to get the insertion event
+ marker_event: The event to process
+ """
+
+ if marker_event.type != EventTypes.MSC2716_MARKER:
+ # Not a marker event
+ return
+
+ if marker_event.rejected_reason is not None:
+ # Rejected event
+ return
+
+ # Skip processing a marker event if the room version doesn't
+ # support it.
+ room_version = await self.store.get_room_version(marker_event.room_id)
+ if not room_version.msc2716_historical:
+ return
+
+ logger.debug("_handle_marker_event: received %s", marker_event)
+
+ insertion_event_id = marker_event.content.get(
+ EventContentFields.MSC2716_MARKER_INSERTION
+ )
+
+ if insertion_event_id is None:
+ # Nothing to retrieve then (invalid marker)
+ return
+
+ logger.debug(
+ "_handle_marker_event: backfilling insertion event %s", insertion_event_id
+ )
+
+ await self._get_events_and_persist(
+ origin,
+ marker_event.room_id,
+ [insertion_event_id],
+ )
+
+ insertion_event = await self.store.get_event(
+ insertion_event_id, allow_none=True
+ )
+ if insertion_event is None:
+ logger.warning(
+ "_handle_marker_event: server %s didn't return insertion event %s for marker %s",
+ origin,
+ insertion_event_id,
+ marker_event.event_id,
+ )
+ return
+
+ logger.debug(
+ "_handle_marker_event: succesfully backfilled insertion event %s from marker event %s",
+ insertion_event,
+ marker_event,
+ )
+
+ await self.store.insert_insertion_extremity(
+ insertion_event_id, marker_event.room_id
+ )
+
+ logger.debug(
+ "_handle_marker_event: insertion extremity added for %s from marker event %s",
+ insertion_event,
+ marker_event,
+ )
+
async def _resync_device(self, sender: str) -> None:
"""We have detected that the device list for the given user may be out
of sync, so we try and resync them.
@@ -1057,9 +1143,19 @@ class FederationHandler(BaseHandler):
async def _maybe_backfill_inner(
self, room_id: str, current_depth: int, limit: int
) -> bool:
- extremities = await self.store.get_oldest_events_with_depth_in_room(room_id)
+ oldest_events_with_depth = (
+ await self.store.get_oldest_event_ids_with_depth_in_room(room_id)
+ )
+ insertion_events_to_be_backfilled = (
+ await self.store.get_insertion_event_backwards_extremities_in_room(room_id)
+ )
+ logger.debug(
+ "_maybe_backfill_inner: extremities oldest_events_with_depth=%s insertion_events_to_be_backfilled=%s",
+ oldest_events_with_depth,
+ insertion_events_to_be_backfilled,
+ )
- if not extremities:
+ if not oldest_events_with_depth and not insertion_events_to_be_backfilled:
logger.debug("Not backfilling as no extremeties found.")
return False
@@ -1089,10 +1185,12 @@ class FederationHandler(BaseHandler):
# state *before* the event, ignoring the special casing certain event
# types have.
- forward_events = await self.store.get_successor_events(list(extremities))
+ forward_event_ids = await self.store.get_successor_events(
+ list(oldest_events_with_depth)
+ )
extremities_events = await self.store.get_events(
- forward_events,
+ forward_event_ids,
redact_behaviour=EventRedactBehaviour.AS_IS,
get_prev_content=False,
)
@@ -1106,10 +1204,19 @@ class FederationHandler(BaseHandler):
redact=False,
check_history_visibility_only=True,
)
+ logger.debug(
+ "_maybe_backfill_inner: filtered_extremities %s", filtered_extremities
+ )
- if not filtered_extremities:
+ if not filtered_extremities and not insertion_events_to_be_backfilled:
return False
+ extremities = {
+ **oldest_events_with_depth,
+ # TODO: insertion_events_to_be_backfilled is currently skipping the filtered_extremities checks
+ **insertion_events_to_be_backfilled,
+ }
+
# Check if we reached a point where we should start backfilling.
sorted_extremeties_tuple = sorted(extremities.items(), key=lambda e: -int(e[1]))
max_depth = sorted_extremeties_tuple[0][1]
diff --git a/synapse/handlers/identity.py b/synapse/handlers/identity.py
index 0961dec5ab..8ffeabacf9 100644
--- a/synapse/handlers/identity.py
+++ b/synapse/handlers/identity.py
@@ -824,6 +824,7 @@ class IdentityHandler(BaseHandler):
room_avatar_url: str,
room_join_rules: str,
room_name: str,
+ room_type: Optional[str],
inviter_display_name: str,
inviter_avatar_url: str,
id_access_token: Optional[str] = None,
@@ -843,6 +844,7 @@ class IdentityHandler(BaseHandler):
notifications.
room_join_rules: The join rules of the email (e.g. "public").
room_name: The m.room.name of the room.
+ room_type: The type of the room from its m.room.create event (e.g "m.space").
inviter_display_name: The current display name of the
inviter.
inviter_avatar_url: The URL of the inviter's avatar.
@@ -869,6 +871,10 @@ class IdentityHandler(BaseHandler):
"sender_display_name": inviter_display_name,
"sender_avatar_url": inviter_avatar_url,
}
+
+ if room_type is not None:
+ invite_config["org.matrix.msc3288.room_type"] = room_type
+
# If a custom web client location is available, include it in the request.
if self._web_client_location:
invite_config["org.matrix.web_client_location"] = self._web_client_location
diff --git a/synapse/handlers/receipts.py b/synapse/handlers/receipts.py
index b9085bbccb..5fd4525700 100644
--- a/synapse/handlers/receipts.py
+++ b/synapse/handlers/receipts.py
@@ -70,7 +70,8 @@ class ReceiptsHandler(BaseHandler):
)
if not is_in_room:
logger.info(
- "Ignoring receipt from %s as we're not in the room",
+ "Ignoring receipt for room %r from server %s as we're not in the room",
+ room_id,
origin,
)
continue
diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index 65ad3efa6a..ba13196218 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -19,7 +19,12 @@ from http import HTTPStatus
from typing import TYPE_CHECKING, Iterable, List, Optional, Set, Tuple
from synapse import types
-from synapse.api.constants import AccountDataTypes, EventTypes, Membership
+from synapse.api.constants import (
+ AccountDataTypes,
+ EventContentFields,
+ EventTypes,
+ Membership,
+)
from synapse.api.errors import (
AuthError,
Codes,
@@ -1237,6 +1242,11 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
if room_name_event:
room_name = room_name_event.content.get("name", "")
+ room_type = None
+ room_create_event = room_state.get((EventTypes.Create, ""))
+ if room_create_event:
+ room_type = room_create_event.content.get(EventContentFields.ROOM_TYPE)
+
room_join_rules = ""
join_rules_event = room_state.get((EventTypes.JoinRules, ""))
if join_rules_event:
@@ -1263,6 +1273,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
room_avatar_url=room_avatar_url,
room_join_rules=room_join_rules,
room_name=room_name,
+ room_type=room_type,
inviter_display_name=inviter_display_name,
inviter_avatar_url=inviter_avatar_url,
id_access_token=id_access_token,
diff --git a/synapse/handlers/space_summary.py b/synapse/handlers/space_summary.py
index 5f7d4602bd..3eb232c83e 100644
--- a/synapse/handlers/space_summary.py
+++ b/synapse/handlers/space_summary.py
@@ -16,7 +16,17 @@ import itertools
import logging
import re
from collections import deque
-from typing import TYPE_CHECKING, Iterable, List, Optional, Sequence, Set, Tuple
+from typing import (
+ TYPE_CHECKING,
+ Collection,
+ Dict,
+ Iterable,
+ List,
+ Optional,
+ Sequence,
+ Set,
+ Tuple,
+)
import attr
@@ -116,20 +126,22 @@ class SpaceSummaryHandler:
max_children = max_rooms_per_space if processed_rooms else None
if is_in_room:
- room, events = await self._summarize_local_room(
+ room_entry = await self._summarize_local_room(
requester, None, room_id, suggested_only, max_children
)
+ events: Collection[JsonDict] = []
+ if room_entry:
+ rooms_result.append(room_entry.room)
+ events = room_entry.children
+
logger.debug(
"Query of local room %s returned events %s",
room_id,
["%s->%s" % (ev["room_id"], ev["state_key"]) for ev in events],
)
-
- if room:
- rooms_result.append(room)
else:
- fed_rooms, fed_events = await self._summarize_remote_room(
+ fed_rooms = await self._summarize_remote_room(
queue_entry,
suggested_only,
max_children,
@@ -141,12 +153,10 @@ class SpaceSummaryHandler:
# user is not permitted see.
#
# Filter the returned results to only what is accessible to the user.
- room_ids = set()
events = []
- for room in fed_rooms:
- fed_room_id = room.get("room_id")
- if not fed_room_id or not isinstance(fed_room_id, str):
- continue
+ for room_entry in fed_rooms:
+ room = room_entry.room
+ fed_room_id = room_entry.room_id
# The room should only be included in the summary if:
# a. the user is in the room;
@@ -189,21 +199,17 @@ class SpaceSummaryHandler:
# The user can see the room, include it!
if include_room:
rooms_result.append(room)
- room_ids.add(fed_room_id)
+ events.extend(room_entry.children)
# All rooms returned don't need visiting again (even if the user
# didn't have access to them).
processed_rooms.add(fed_room_id)
- for event in fed_events:
- if event.get("room_id") in room_ids:
- events.append(event)
-
logger.debug(
"Query of %s returned rooms %s, events %s",
room_id,
- [room.get("room_id") for room in fed_rooms],
- ["%s->%s" % (ev["room_id"], ev["state_key"]) for ev in fed_events],
+ [room_entry.room.get("room_id") for room_entry in fed_rooms],
+ ["%s->%s" % (ev["room_id"], ev["state_key"]) for ev in events],
)
# the room we queried may or may not have been returned, but don't process
@@ -283,20 +289,20 @@ class SpaceSummaryHandler:
# already done this room
continue
- logger.debug("Processing room %s", room_id)
-
- room, events = await self._summarize_local_room(
+ room_entry = await self._summarize_local_room(
None, origin, room_id, suggested_only, max_rooms_per_space
)
processed_rooms.add(room_id)
- if room:
- rooms_result.append(room)
- events_result.extend(events)
+ if room_entry:
+ rooms_result.append(room_entry.room)
+ events_result.extend(room_entry.children)
- # add any children to the queue
- room_queue.extend(edge_event["state_key"] for edge_event in events)
+ # add any children to the queue
+ room_queue.extend(
+ edge_event["state_key"] for edge_event in room_entry.children
+ )
return {"rooms": rooms_result, "events": events_result}
@@ -307,7 +313,7 @@ class SpaceSummaryHandler:
room_id: str,
suggested_only: bool,
max_children: Optional[int],
- ) -> Tuple[Optional[JsonDict], Sequence[JsonDict]]:
+ ) -> Optional["_RoomEntry"]:
"""
Generate a room entry and a list of event entries for a given room.
@@ -326,21 +332,16 @@ class SpaceSummaryHandler:
to a server-set limit.
Returns:
- A tuple of:
- The room information, if the room should be returned to the
- user. None, otherwise.
-
- An iterable of the sorted children events. This may be limited
- to a maximum size or may include all children.
+ A room entry if the room should be returned. None, otherwise.
"""
if not await self._is_room_accessible(room_id, requester, origin):
- return None, ()
+ return None
room_entry = await self._build_room_entry(room_id)
# If the room is not a space, return just the room information.
if room_entry.get("room_type") != RoomTypes.SPACE:
- return room_entry, ()
+ return _RoomEntry(room_id, room_entry)
# Otherwise, look for child rooms/spaces.
child_events = await self._get_child_events(room_id)
@@ -363,7 +364,7 @@ class SpaceSummaryHandler:
)
)
- return room_entry, events_result
+ return _RoomEntry(room_id, room_entry, events_result)
async def _summarize_remote_room(
self,
@@ -371,7 +372,7 @@ class SpaceSummaryHandler:
suggested_only: bool,
max_children: Optional[int],
exclude_rooms: Iterable[str],
- ) -> Tuple[Sequence[JsonDict], Sequence[JsonDict]]:
+ ) -> Iterable["_RoomEntry"]:
"""
Request room entries and a list of event entries for a given room by querying a remote server.
@@ -386,11 +387,7 @@ class SpaceSummaryHandler:
Rooms IDs which do not need to be summarized.
Returns:
- A tuple of:
- An iterable of rooms.
-
- An iterable of the sorted children events. This may be limited
- to a maximum size or may include all children.
+ An iterable of room entries.
"""
room_id = room.room_id
logger.info("Requesting summary for %s via %s", room_id, room.via)
@@ -414,11 +411,30 @@ class SpaceSummaryHandler:
e,
exc_info=logger.isEnabledFor(logging.DEBUG),
)
- return (), ()
+ return ()
+
+ # Group the events by their room.
+ children_by_room: Dict[str, List[JsonDict]] = {}
+ for ev in res.events:
+ if ev.event_type == EventTypes.SpaceChild:
+ children_by_room.setdefault(ev.room_id, []).append(ev.data)
+
+ # Generate the final results.
+ results = []
+ for fed_room in res.rooms:
+ fed_room_id = fed_room.get("room_id")
+ if not fed_room_id or not isinstance(fed_room_id, str):
+ continue
- return res.rooms, tuple(
- ev.data for ev in res.events if ev.event_type == EventTypes.SpaceChild
- )
+ results.append(
+ _RoomEntry(
+ fed_room_id,
+ fed_room,
+ children_by_room.get(fed_room_id, []),
+ )
+ )
+
+ return results
async def _is_room_accessible(
self, room_id: str, requester: Optional[str], origin: Optional[str]
@@ -606,10 +622,21 @@ class SpaceSummaryHandler:
return sorted(filter(_has_valid_via, events), key=_child_events_comparison_key)
-@attr.s(frozen=True, slots=True)
+@attr.s(frozen=True, slots=True, auto_attribs=True)
class _RoomQueueEntry:
- room_id = attr.ib(type=str)
- via = attr.ib(type=Sequence[str])
+ room_id: str
+ via: Sequence[str]
+
+
+@attr.s(frozen=True, slots=True, auto_attribs=True)
+class _RoomEntry:
+ room_id: str
+ # The room summary for this room.
+ room: JsonDict
+ # An iterable of the sorted, stripped children events for children of this room.
+ #
+ # This may not include all children.
+ children: Collection[JsonDict] = ()
def _has_valid_via(e: EventBase) -> bool:
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index f30bfcc93c..590642f510 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -269,14 +269,22 @@ class SyncHandler:
self.presence_handler = hs.get_presence_handler()
self.event_sources = hs.get_event_sources()
self.clock = hs.get_clock()
- self.response_cache: ResponseCache[SyncRequestKey] = ResponseCache(
- hs.get_clock(), "sync"
- )
self.state = hs.get_state_handler()
self.auth = hs.get_auth()
self.storage = hs.get_storage()
self.state_store = self.storage.state
+ # TODO: flush cache entries on subsequent sync request.
+ # Once we get the next /sync request (ie, one with the same access token
+ # that sets 'since' to 'next_batch'), we know that device won't need a
+ # cached result any more, and we could flush the entry from the cache to save
+ # memory.
+ self.response_cache: ResponseCache[SyncRequestKey] = ResponseCache(
+ hs.get_clock(),
+ "sync",
+ timeout_ms=hs.config.caches.sync_response_cache_duration,
+ )
+
# ExpiringCache((User, Device)) -> LruCache(user_id => event_id)
self.lazy_loaded_members_cache: ExpiringCache[
Tuple[str, Optional[str]], LruCache[str, str]
diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index 0cb651a400..a97c448595 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -335,7 +335,8 @@ class TypingWriterHandler(FollowerTypingHandler):
)
if not is_in_room:
logger.info(
- "Ignoring typing update from %s as we're not in the room",
+ "Ignoring typing update for room %r from server %s as we're not in the room",
+ room_id,
origin,
)
return
|
