12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124 |
- # Copyright 2014-2016 OpenMarket Ltd
- # Copyright 2018-2019 New Vector Ltd
- # Copyright 2019 The Matrix.org Foundation C.I.C.
- #
- # Licensed under the Apache License, Version 2.0 (the "License");
- # you may not use this file except in compliance with the License.
- # You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- import itertools
- import logging
- from collections import deque
- from typing import (
- TYPE_CHECKING,
- Any,
- Awaitable,
- Callable,
- Collection,
- Deque,
- Dict,
- Generator,
- Generic,
- Iterable,
- List,
- Optional,
- Set,
- Tuple,
- TypeVar,
- )
- import attr
- from prometheus_client import Counter, Histogram
- from twisted.internet import defer
- from synapse.api.constants import EventTypes, Membership
- from synapse.events import EventBase
- from synapse.events.snapshot import EventContext
- from synapse.logging import opentracing
- from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable
- from synapse.metrics.background_process_metrics import run_as_background_process
- from synapse.storage.databases import Databases
- from synapse.storage.databases.main.events import DeltaState
- from synapse.storage.databases.main.events_worker import EventRedactBehaviour
- from synapse.types import (
- PersistedEventPosition,
- RoomStreamToken,
- StateMap,
- get_domain_from_id,
- )
- from synapse.util.async_helpers import ObservableDeferred, yieldable_gather_results
- from synapse.util.metrics import Measure
- if TYPE_CHECKING:
- from synapse.server import HomeServer
- logger = logging.getLogger(__name__)
- # The number of times we are recalculating the current state
- state_delta_counter = Counter("synapse_storage_events_state_delta", "")
- # The number of times we are recalculating state when there is only a
- # single forward extremity
- state_delta_single_event_counter = Counter(
- "synapse_storage_events_state_delta_single_event", ""
- )
- # The number of times we are reculating state when we could have resonably
- # calculated the delta when we calculated the state for an event we were
- # persisting.
- state_delta_reuse_delta_counter = Counter(
- "synapse_storage_events_state_delta_reuse_delta", ""
- )
- # The number of forward extremities for each new event.
- forward_extremities_counter = Histogram(
- "synapse_storage_events_forward_extremities_persisted",
- "Number of forward extremities for each new event",
- buckets=(1, 2, 3, 5, 7, 10, 15, 20, 50, 100, 200, 500, "+Inf"),
- )
- # The number of stale forward extremities for each new event. Stale extremities
- # are those that were in the previous set of extremities as well as the new.
- stale_forward_extremities_counter = Histogram(
- "synapse_storage_events_stale_forward_extremities_persisted",
- "Number of unchanged forward extremities for each new event",
- buckets=(0, 1, 2, 3, 5, 7, 10, 15, 20, 50, 100, 200, 500, "+Inf"),
- )
- state_resolutions_during_persistence = Counter(
- "synapse_storage_events_state_resolutions_during_persistence",
- "Number of times we had to do state res to calculate new current state",
- )
- potential_times_prune_extremities = Counter(
- "synapse_storage_events_potential_times_prune_extremities",
- "Number of times we might be able to prune extremities",
- )
- times_pruned_extremities = Counter(
- "synapse_storage_events_times_pruned_extremities",
- "Number of times we were actually be able to prune extremities",
- )
- @attr.s(auto_attribs=True, slots=True)
- class _EventPersistQueueItem:
- events_and_contexts: List[Tuple[EventBase, EventContext]]
- backfilled: bool
- deferred: ObservableDeferred
- parent_opentracing_span_contexts: List = attr.ib(factory=list)
- """A list of opentracing spans waiting for this batch"""
- opentracing_span_context: Any = None
- """The opentracing span under which the persistence actually happened"""
- _PersistResult = TypeVar("_PersistResult")
- class _EventPeristenceQueue(Generic[_PersistResult]):
- """Queues up events so that they can be persisted in bulk with only one
- concurrent transaction per room.
- """
- def __init__(
- self,
- per_item_callback: Callable[
- [List[Tuple[EventBase, EventContext]], bool],
- Awaitable[_PersistResult],
- ],
- ):
- """Create a new event persistence queue
- The per_item_callback will be called for each item added via add_to_queue,
- and its result will be returned via the Deferreds returned from add_to_queue.
- """
- self._event_persist_queues: Dict[str, Deque[_EventPersistQueueItem]] = {}
- self._currently_persisting_rooms: Set[str] = set()
- self._per_item_callback = per_item_callback
- async def add_to_queue(
- self,
- room_id: str,
- events_and_contexts: Iterable[Tuple[EventBase, EventContext]],
- backfilled: bool,
- ) -> _PersistResult:
- """Add events to the queue, with the given persist_event options.
- If we are not already processing events in this room, starts off a background
- process to to so, calling the per_item_callback for each item.
- Args:
- room_id (str):
- events_and_contexts (list[(EventBase, EventContext)]):
- backfilled (bool):
- Returns:
- the result returned by the `_per_item_callback` passed to
- `__init__`.
- """
- queue = self._event_persist_queues.setdefault(room_id, deque())
- # if the last item in the queue has the same `backfilled` setting,
- # we can just add these new events to that item.
- if queue and queue[-1].backfilled == backfilled:
- end_item = queue[-1]
- else:
- # need to make a new queue item
- deferred: ObservableDeferred[_PersistResult] = ObservableDeferred(
- defer.Deferred(), consumeErrors=True
- )
- end_item = _EventPersistQueueItem(
- events_and_contexts=[],
- backfilled=backfilled,
- deferred=deferred,
- )
- queue.append(end_item)
- # add our events to the queue item
- end_item.events_and_contexts.extend(events_and_contexts)
- # also add our active opentracing span to the item so that we get a link back
- span = opentracing.active_span()
- if span:
- end_item.parent_opentracing_span_contexts.append(span.context)
- # start a processor for the queue, if there isn't one already
- self._handle_queue(room_id)
- # wait for the queue item to complete
- res = await make_deferred_yieldable(end_item.deferred.observe())
- # add another opentracing span which links to the persist trace.
- with opentracing.start_active_span_follows_from(
- "persist_event_batch_complete", (end_item.opentracing_span_context,)
- ):
- pass
- return res
- def _handle_queue(self, room_id: str) -> None:
- """Attempts to handle the queue for a room if not already being handled.
- The queue's callback will be invoked with for each item in the queue,
- of type _EventPersistQueueItem. The per_item_callback will continuously
- be called with new items, unless the queue becomes empty. The return
- value of the function will be given to the deferreds waiting on the item,
- exceptions will be passed to the deferreds as well.
- This function should therefore be called whenever anything is added
- to the queue.
- If another callback is currently handling the queue then it will not be
- invoked.
- """
- if room_id in self._currently_persisting_rooms:
- return
- self._currently_persisting_rooms.add(room_id)
- async def handle_queue_loop() -> None:
- try:
- queue = self._get_drainining_queue(room_id)
- for item in queue:
- try:
- with opentracing.start_active_span_follows_from(
- "persist_event_batch",
- item.parent_opentracing_span_contexts,
- inherit_force_tracing=True,
- ) as scope:
- if scope:
- item.opentracing_span_context = scope.span.context
- ret = await self._per_item_callback(
- item.events_and_contexts, item.backfilled
- )
- except Exception:
- with PreserveLoggingContext():
- item.deferred.errback()
- else:
- with PreserveLoggingContext():
- item.deferred.callback(ret)
- finally:
- remaining_queue = self._event_persist_queues.pop(room_id, None)
- if remaining_queue:
- self._event_persist_queues[room_id] = remaining_queue
- self._currently_persisting_rooms.discard(room_id)
- # set handle_queue_loop off in the background
- run_as_background_process("persist_events", handle_queue_loop)
- def _get_drainining_queue(
- self, room_id: str
- ) -> Generator[_EventPersistQueueItem, None, None]:
- queue = self._event_persist_queues.setdefault(room_id, deque())
- try:
- while True:
- yield queue.popleft()
- except IndexError:
- # Queue has been drained.
- pass
- class EventsPersistenceStorageController:
- """High level interface for handling persisting newly received events.
- Takes care of batching up events by room, and calculating the necessary
- current state and forward extremity changes.
- """
- def __init__(self, hs: "HomeServer", stores: Databases):
- # We ultimately want to split out the state store from the main store,
- # so we use separate variables here even though they point to the same
- # store for now.
- self.main_store = stores.main
- self.state_store = stores.state
- assert stores.persist_events
- self.persist_events_store = stores.persist_events
- self._clock = hs.get_clock()
- self._instance_name = hs.get_instance_name()
- self.is_mine_id = hs.is_mine_id
- self._event_persist_queue = _EventPeristenceQueue(self._persist_event_batch)
- self._state_resolution_handler = hs.get_state_resolution_handler()
- @opentracing.trace
- async def persist_events(
- self,
- events_and_contexts: Iterable[Tuple[EventBase, EventContext]],
- backfilled: bool = False,
- ) -> Tuple[List[EventBase], RoomStreamToken]:
- """
- Write events to the database
- Args:
- events_and_contexts: list of tuples of (event, context)
- backfilled: Whether the results are retrieved from federation
- via backfill or not. Used to determine if they're "new" events
- which might update the current state etc.
- Returns:
- List of events persisted, the current position room stream position.
- The list of events persisted may not be the same as those passed in
- if they were deduplicated due to an event already existing that
- matched the transaction ID; the existing event is returned in such
- a case.
- """
- partitioned: Dict[str, List[Tuple[EventBase, EventContext]]] = {}
- for event, ctx in events_and_contexts:
- partitioned.setdefault(event.room_id, []).append((event, ctx))
- async def enqueue(
- item: Tuple[str, List[Tuple[EventBase, EventContext]]]
- ) -> Dict[str, str]:
- room_id, evs_ctxs = item
- return await self._event_persist_queue.add_to_queue(
- room_id, evs_ctxs, backfilled=backfilled
- )
- ret_vals = await yieldable_gather_results(enqueue, partitioned.items())
- # Each call to add_to_queue returns a map from event ID to existing event ID if
- # the event was deduplicated. (The dict may also include other entries if
- # the event was persisted in a batch with other events).
- #
- # Since we use `yieldable_gather_results` we need to merge the returned list
- # of dicts into one.
- replaced_events: Dict[str, str] = {}
- for d in ret_vals:
- replaced_events.update(d)
- events = []
- for event, _ in events_and_contexts:
- existing_event_id = replaced_events.get(event.event_id)
- if existing_event_id:
- events.append(await self.main_store.get_event(existing_event_id))
- else:
- events.append(event)
- return (
- events,
- self.main_store.get_room_max_token(),
- )
- @opentracing.trace
- async def persist_event(
- self, event: EventBase, context: EventContext, backfilled: bool = False
- ) -> Tuple[EventBase, PersistedEventPosition, RoomStreamToken]:
- """
- Returns:
- The event, stream ordering of `event`, and the stream ordering of the
- latest persisted event. The returned event may not match the given
- event if it was deduplicated due to an existing event matching the
- transaction ID.
- """
- # add_to_queue returns a map from event ID to existing event ID if the
- # event was deduplicated. (The dict may also include other entries if
- # the event was persisted in a batch with other events.)
- replaced_events = await self._event_persist_queue.add_to_queue(
- event.room_id, [(event, context)], backfilled=backfilled
- )
- replaced_event = replaced_events.get(event.event_id)
- if replaced_event:
- event = await self.main_store.get_event(replaced_event)
- event_stream_id = event.internal_metadata.stream_ordering
- # stream ordering should have been assigned by now
- assert event_stream_id
- pos = PersistedEventPosition(self._instance_name, event_stream_id)
- return event, pos, self.main_store.get_room_max_token()
- async def update_current_state(self, room_id: str) -> None:
- """Recalculate the current state for a room, and persist it"""
- state = await self._calculate_current_state(room_id)
- delta = await self._calculate_state_delta(room_id, state)
- # TODO(faster_joins): get a real stream ordering, to make this work correctly
- # across workers.
- #
- # TODO(faster_joins): this can race against event persistence, in which case we
- # will end up with incorrect state. Perhaps we should make this a job we
- # farm out to the event persister, somehow.
- stream_id = self.main_store.get_room_max_stream_ordering()
- await self.persist_events_store.update_current_state(room_id, delta, stream_id)
- async def _calculate_current_state(self, room_id: str) -> StateMap[str]:
- """Calculate the current state of a room, based on the forward extremities
- Args:
- room_id: room for which to calculate current state
- Returns:
- map from (type, state_key) to event id for the current state in the room
- """
- latest_event_ids = await self.main_store.get_latest_event_ids_in_room(room_id)
- state_groups = set(
- (
- await self.main_store._get_state_group_for_events(latest_event_ids)
- ).values()
- )
- state_maps_by_state_group = await self.state_store._get_state_for_groups(
- state_groups
- )
- if len(state_groups) == 1:
- # If there is only one state group, then we know what the current
- # state is.
- return state_maps_by_state_group[state_groups.pop()]
- # Ok, we need to defer to the state handler to resolve our state sets.
- logger.debug("calling resolve_state_groups from preserve_events")
- # Avoid a circular import.
- from synapse.state import StateResolutionStore
- room_version = await self.main_store.get_room_version_id(room_id)
- res = await self._state_resolution_handler.resolve_state_groups(
- room_id,
- room_version,
- state_maps_by_state_group,
- event_map=None,
- state_res_store=StateResolutionStore(self.main_store),
- )
- return res.state
- async def _persist_event_batch(
- self,
- events_and_contexts: List[Tuple[EventBase, EventContext]],
- backfilled: bool = False,
- ) -> Dict[str, str]:
- """Callback for the _event_persist_queue
- Calculates the change to current state and forward extremities, and
- persists the given events and with those updates.
- Returns:
- A dictionary of event ID to event ID we didn't persist as we already
- had another event persisted with the same TXN ID.
- """
- replaced_events: Dict[str, str] = {}
- if not events_and_contexts:
- return replaced_events
- # Check if any of the events have a transaction ID that has already been
- # persisted, and if so we don't persist it again.
- #
- # We should have checked this a long time before we get here, but it's
- # possible that different send event requests race in such a way that
- # they both pass the earlier checks. Checking here isn't racey as we can
- # have only one `_persist_events` per room being called at a time.
- replaced_events = await self.main_store.get_already_persisted_events(
- (event for event, _ in events_and_contexts)
- )
- if replaced_events:
- events_and_contexts = [
- (e, ctx)
- for e, ctx in events_and_contexts
- if e.event_id not in replaced_events
- ]
- if not events_and_contexts:
- return replaced_events
- chunks = [
- events_and_contexts[x : x + 100]
- for x in range(0, len(events_and_contexts), 100)
- ]
- for chunk in chunks:
- # We can't easily parallelize these since different chunks
- # might contain the same event. :(
- # NB: Assumes that we are only persisting events for one room
- # at a time.
- # map room_id->set[event_ids] giving the new forward
- # extremities in each room
- new_forward_extremities: Dict[str, Set[str]] = {}
- # map room_id->(to_delete, to_insert) where to_delete is a list
- # of type/state keys to remove from current state, and to_insert
- # is a map (type,key)->event_id giving the state delta in each
- # room
- state_delta_for_room: Dict[str, DeltaState] = {}
- # Set of remote users which were in rooms the server has left. We
- # should check if we still share any rooms and if not we mark their
- # device lists as stale.
- potentially_left_users: Set[str] = set()
- if not backfilled:
- with Measure(self._clock, "_calculate_state_and_extrem"):
- # Work out the new "current state" for each room.
- # We do this by working out what the new extremities are and then
- # calculating the state from that.
- events_by_room: Dict[str, List[Tuple[EventBase, EventContext]]] = {}
- for event, context in chunk:
- events_by_room.setdefault(event.room_id, []).append(
- (event, context)
- )
- for room_id, ev_ctx_rm in events_by_room.items():
- latest_event_ids = set(
- await self.main_store.get_latest_event_ids_in_room(room_id)
- )
- new_latest_event_ids = await self._calculate_new_extremities(
- room_id, ev_ctx_rm, latest_event_ids
- )
- if new_latest_event_ids == latest_event_ids:
- # No change in extremities, so no change in state
- continue
- # there should always be at least one forward extremity.
- # (except during the initial persistence of the send_join
- # results, in which case there will be no existing
- # extremities, so we'll `continue` above and skip this bit.)
- assert new_latest_event_ids, "No forward extremities left!"
- new_forward_extremities[room_id] = new_latest_event_ids
- len_1 = (
- len(latest_event_ids) == 1
- and len(new_latest_event_ids) == 1
- )
- if len_1:
- all_single_prev_not_state = all(
- len(event.prev_event_ids()) == 1
- and not event.is_state()
- for event, ctx in ev_ctx_rm
- )
- # Don't bother calculating state if they're just
- # a long chain of single ancestor non-state events.
- if all_single_prev_not_state:
- continue
- state_delta_counter.inc()
- if len(new_latest_event_ids) == 1:
- state_delta_single_event_counter.inc()
- # This is a fairly handwavey check to see if we could
- # have guessed what the delta would have been when
- # processing one of these events.
- # What we're interested in is if the latest extremities
- # were the same when we created the event as they are
- # now. When this server creates a new event (as opposed
- # to receiving it over federation) it will use the
- # forward extremities as the prev_events, so we can
- # guess this by looking at the prev_events and checking
- # if they match the current forward extremities.
- for ev, _ in ev_ctx_rm:
- prev_event_ids = set(ev.prev_event_ids())
- if latest_event_ids == prev_event_ids:
- state_delta_reuse_delta_counter.inc()
- break
- logger.debug("Calculating state delta for room %s", room_id)
- with Measure(
- self._clock, "persist_events.get_new_state_after_events"
- ):
- res = await self._get_new_state_after_events(
- room_id,
- ev_ctx_rm,
- latest_event_ids,
- new_latest_event_ids,
- )
- current_state, delta_ids, new_latest_event_ids = res
- # there should always be at least one forward extremity.
- # (except during the initial persistence of the send_join
- # results, in which case there will be no existing
- # extremities, so we'll `continue` above and skip this bit.)
- assert new_latest_event_ids, "No forward extremities left!"
- new_forward_extremities[room_id] = new_latest_event_ids
- # If either are not None then there has been a change,
- # and we need to work out the delta (or use that
- # given)
- delta = None
- if delta_ids is not None:
- # If there is a delta we know that we've
- # only added or replaced state, never
- # removed keys entirely.
- delta = DeltaState([], delta_ids)
- elif current_state is not None:
- with Measure(
- self._clock, "persist_events.calculate_state_delta"
- ):
- delta = await self._calculate_state_delta(
- room_id, current_state
- )
- if delta:
- # If we have a change of state then lets check
- # whether we're actually still a member of the room,
- # or if our last user left. If we're no longer in
- # the room then we delete the current state and
- # extremities.
- is_still_joined = await self._is_server_still_joined(
- room_id,
- ev_ctx_rm,
- delta,
- current_state,
- potentially_left_users,
- )
- if not is_still_joined:
- logger.info("Server no longer in room %s", room_id)
- latest_event_ids = set()
- current_state = {}
- delta.no_longer_in_room = True
- state_delta_for_room[room_id] = delta
- await self.persist_events_store._persist_events_and_state_updates(
- chunk,
- state_delta_for_room=state_delta_for_room,
- new_forward_extremities=new_forward_extremities,
- use_negative_stream_ordering=backfilled,
- inhibit_local_membership_updates=backfilled,
- )
- await self._handle_potentially_left_users(potentially_left_users)
- return replaced_events
- async def _calculate_new_extremities(
- self,
- room_id: str,
- event_contexts: List[Tuple[EventBase, EventContext]],
- latest_event_ids: Collection[str],
- ) -> Set[str]:
- """Calculates the new forward extremities for a room given events to
- persist.
- Assumes that we are only persisting events for one room at a time.
- """
- # we're only interested in new events which aren't outliers and which aren't
- # being rejected.
- new_events = [
- event
- for event, ctx in event_contexts
- if not event.internal_metadata.is_outlier()
- and not ctx.rejected
- and not event.internal_metadata.is_soft_failed()
- ]
- latest_event_ids = set(latest_event_ids)
- # start with the existing forward extremities
- result = set(latest_event_ids)
- # add all the new events to the list
- result.update(event.event_id for event in new_events)
- # Now remove all events which are prev_events of any of the new events
- result.difference_update(
- e_id for event in new_events for e_id in event.prev_event_ids()
- )
- # Remove any events which are prev_events of any existing events.
- existing_prevs: Collection[
- str
- ] = await self.persist_events_store._get_events_which_are_prevs(result)
- result.difference_update(existing_prevs)
- # Finally handle the case where the new events have soft-failed prev
- # events. If they do we need to remove them and their prev events,
- # otherwise we end up with dangling extremities.
- existing_prevs = await self.persist_events_store._get_prevs_before_rejected(
- e_id for event in new_events for e_id in event.prev_event_ids()
- )
- result.difference_update(existing_prevs)
- # We only update metrics for events that change forward extremities
- # (e.g. we ignore backfill/outliers/etc)
- if result != latest_event_ids:
- forward_extremities_counter.observe(len(result))
- stale = latest_event_ids & result
- stale_forward_extremities_counter.observe(len(stale))
- return result
- async def _get_new_state_after_events(
- self,
- room_id: str,
- events_context: List[Tuple[EventBase, EventContext]],
- old_latest_event_ids: Set[str],
- new_latest_event_ids: Set[str],
- ) -> Tuple[Optional[StateMap[str]], Optional[StateMap[str]], Set[str]]:
- """Calculate the current state dict after adding some new events to
- a room
- Args:
- room_id:
- room to which the events are being added. Used for logging etc
- events_context:
- events and contexts which are being added to the room
- old_latest_event_ids:
- the old forward extremities for the room.
- new_latest_event_ids :
- the new forward extremities for the room.
- Returns:
- Returns a tuple of two state maps and a set of new forward
- extremities.
- The first state map is the full new current state and the second
- is the delta to the existing current state. If both are None then
- there has been no change. Either or neither can be None if there
- has been a change.
- The function may prune some old entries from the set of new
- forward extremities if it's safe to do so.
- If there has been a change then we only return the delta if its
- already been calculated. Conversely if we do know the delta then
- the new current state is only returned if we've already calculated
- it.
- """
- # Map from (prev state group, new state group) -> delta state dict
- state_group_deltas = {}
- for ev, ctx in events_context:
- if ctx.state_group is None:
- # This should only happen for outlier events.
- if not ev.internal_metadata.is_outlier():
- raise Exception(
- "Context for new event %s has no state "
- "group" % (ev.event_id,)
- )
- continue
- if ctx.prev_group:
- state_group_deltas[(ctx.prev_group, ctx.state_group)] = ctx.delta_ids
- # We need to map the event_ids to their state groups. First, let's
- # check if the event is one we're persisting, in which case we can
- # pull the state group from its context.
- # Otherwise we need to pull the state group from the database.
- # Set of events we need to fetch groups for. (We know none of the old
- # extremities are going to be in events_context).
- missing_event_ids = set(old_latest_event_ids)
- event_id_to_state_group = {}
- for event_id in new_latest_event_ids:
- # First search in the list of new events we're adding.
- for ev, ctx in events_context:
- if event_id == ev.event_id and ctx.state_group is not None:
- event_id_to_state_group[event_id] = ctx.state_group
- break
- else:
- # If we couldn't find it, then we'll need to pull
- # the state from the database
- missing_event_ids.add(event_id)
- if missing_event_ids:
- # Now pull out the state groups for any missing events from DB
- event_to_groups = await self.main_store._get_state_group_for_events(
- missing_event_ids
- )
- event_id_to_state_group.update(event_to_groups)
- # State groups of old_latest_event_ids
- old_state_groups = {
- event_id_to_state_group[evid] for evid in old_latest_event_ids
- }
- # State groups of new_latest_event_ids
- new_state_groups = {
- event_id_to_state_group[evid] for evid in new_latest_event_ids
- }
- # If they old and new groups are the same then we don't need to do
- # anything.
- if old_state_groups == new_state_groups:
- return None, None, new_latest_event_ids
- if len(new_state_groups) == 1 and len(old_state_groups) == 1:
- # If we're going from one state group to another, lets check if
- # we have a delta for that transition. If we do then we can just
- # return that.
- new_state_group = next(iter(new_state_groups))
- old_state_group = next(iter(old_state_groups))
- delta_ids = state_group_deltas.get((old_state_group, new_state_group), None)
- if delta_ids is not None:
- # We have a delta from the existing to new current state,
- # so lets just return that.
- return None, delta_ids, new_latest_event_ids
- # Now that we have calculated new_state_groups we need to get
- # their state IDs so we can resolve to a single state set.
- state_groups_map = await self.state_store._get_state_for_groups(
- new_state_groups
- )
- if len(new_state_groups) == 1:
- # If there is only one state group, then we know what the current
- # state is.
- return state_groups_map[new_state_groups.pop()], None, new_latest_event_ids
- # Ok, we need to defer to the state handler to resolve our state sets.
- state_groups = {sg: state_groups_map[sg] for sg in new_state_groups}
- events_map = {ev.event_id: ev for ev, _ in events_context}
- # We need to get the room version, which is in the create event.
- # Normally that'd be in the database, but its also possible that we're
- # currently trying to persist it.
- room_version = None
- for ev, _ in events_context:
- if ev.type == EventTypes.Create and ev.state_key == "":
- room_version = ev.content.get("room_version", "1")
- break
- if not room_version:
- room_version = await self.main_store.get_room_version_id(room_id)
- logger.debug("calling resolve_state_groups from preserve_events")
- # Avoid a circular import.
- from synapse.state import StateResolutionStore
- res = await self._state_resolution_handler.resolve_state_groups(
- room_id,
- room_version,
- state_groups,
- events_map,
- state_res_store=StateResolutionStore(self.main_store),
- )
- state_resolutions_during_persistence.inc()
- # If the returned state matches the state group of one of the new
- # forward extremities then we check if we are able to prune some state
- # extremities.
- if res.state_group and res.state_group in new_state_groups:
- new_latest_event_ids = await self._prune_extremities(
- room_id,
- new_latest_event_ids,
- res.state_group,
- event_id_to_state_group,
- events_context,
- )
- return res.state, None, new_latest_event_ids
- async def _prune_extremities(
- self,
- room_id: str,
- new_latest_event_ids: Set[str],
- resolved_state_group: int,
- event_id_to_state_group: Dict[str, int],
- events_context: List[Tuple[EventBase, EventContext]],
- ) -> Set[str]:
- """See if we can prune any of the extremities after calculating the
- resolved state.
- """
- potential_times_prune_extremities.inc()
- # We keep all the extremities that have the same state group, and
- # see if we can drop the others.
- new_new_extrems = {
- e
- for e in new_latest_event_ids
- if event_id_to_state_group[e] == resolved_state_group
- }
- dropped_extrems = set(new_latest_event_ids) - new_new_extrems
- logger.debug("Might drop extremities: %s", dropped_extrems)
- # We only drop events from the extremities list if:
- # 1. we're not currently persisting them;
- # 2. they're not our own events (or are dummy events); and
- # 3. they're either:
- # 1. over N hours old and more than N events ago (we use depth to
- # calculate); or
- # 2. we are persisting an event from the same domain and more than
- # M events ago.
- #
- # The idea is that we don't want to drop events that are "legitimate"
- # extremities (that we would want to include as prev events), only
- # "stuck" extremities that are e.g. due to a gap in the graph.
- #
- # Note that we either drop all of them or none of them. If we only drop
- # some of the events we don't know if state res would come to the same
- # conclusion.
- for ev, _ in events_context:
- if ev.event_id in dropped_extrems:
- logger.debug(
- "Not dropping extremities: %s is being persisted", ev.event_id
- )
- return new_latest_event_ids
- dropped_events = await self.main_store.get_events(
- dropped_extrems,
- allow_rejected=True,
- redact_behaviour=EventRedactBehaviour.as_is,
- )
- new_senders = {get_domain_from_id(e.sender) for e, _ in events_context}
- one_day_ago = self._clock.time_msec() - 24 * 60 * 60 * 1000
- current_depth = max(e.depth for e, _ in events_context)
- for event in dropped_events.values():
- # If the event is a local dummy event then we should check it
- # doesn't reference any local events, as we want to reference those
- # if we send any new events.
- #
- # Note we do this recursively to handle the case where a dummy event
- # references a dummy event that only references remote events.
- #
- # Ideally we'd figure out a way of still being able to drop old
- # dummy events that reference local events, but this is good enough
- # as a first cut.
- events_to_check: Collection[EventBase] = [event]
- while events_to_check:
- new_events: Set[str] = set()
- for event_to_check in events_to_check:
- if self.is_mine_id(event_to_check.sender):
- if event_to_check.type != EventTypes.Dummy:
- logger.debug("Not dropping own event")
- return new_latest_event_ids
- new_events.update(event_to_check.prev_event_ids())
- prev_events = await self.main_store.get_events(
- new_events,
- allow_rejected=True,
- redact_behaviour=EventRedactBehaviour.as_is,
- )
- events_to_check = prev_events.values()
- if (
- event.origin_server_ts < one_day_ago
- and event.depth < current_depth - 100
- ):
- continue
- # We can be less conservative about dropping extremities from the
- # same domain, though we do want to wait a little bit (otherwise
- # we'll immediately remove all extremities from a given server).
- if (
- get_domain_from_id(event.sender) in new_senders
- and event.depth < current_depth - 20
- ):
- continue
- logger.debug(
- "Not dropping as too new and not in new_senders: %s",
- new_senders,
- )
- return new_latest_event_ids
- times_pruned_extremities.inc()
- logger.info(
- "Pruning forward extremities in room %s: from %s -> %s",
- room_id,
- new_latest_event_ids,
- new_new_extrems,
- )
- return new_new_extrems
- async def _calculate_state_delta(
- self, room_id: str, current_state: StateMap[str]
- ) -> DeltaState:
- """Calculate the new state deltas for a room.
- Assumes that we are only persisting events for one room at a time.
- """
- existing_state = await self.main_store.get_partial_current_state_ids(room_id)
- to_delete = [key for key in existing_state if key not in current_state]
- to_insert = {
- key: ev_id
- for key, ev_id in current_state.items()
- if ev_id != existing_state.get(key)
- }
- return DeltaState(to_delete=to_delete, to_insert=to_insert)
- async def _is_server_still_joined(
- self,
- room_id: str,
- ev_ctx_rm: List[Tuple[EventBase, EventContext]],
- delta: DeltaState,
- current_state: Optional[StateMap[str]],
- potentially_left_users: Set[str],
- ) -> bool:
- """Check if the server will still be joined after the given events have
- been persised.
- Args:
- room_id
- ev_ctx_rm
- delta: The delta of current state between what is in the database
- and what the new current state will be.
- current_state: The new current state if it already been calculated,
- otherwise None.
- potentially_left_users: If the server has left the room, then joined
- remote users will be added to this set to indicate that the
- server may no longer be sharing a room with them.
- """
- if not any(
- self.is_mine_id(state_key)
- for typ, state_key in itertools.chain(delta.to_delete, delta.to_insert)
- if typ == EventTypes.Member
- ):
- # There have been no changes to membership of our users, so nothing
- # has changed and we assume we're still in the room.
- return True
- # Check if any of the given events are a local join that appear in the
- # current state
- events_to_check = [] # Event IDs that aren't an event we're persisting
- for (typ, state_key), event_id in delta.to_insert.items():
- if typ != EventTypes.Member or not self.is_mine_id(state_key):
- continue
- for event, _ in ev_ctx_rm:
- if event_id == event.event_id:
- if event.membership == Membership.JOIN:
- return True
- # The event is not in `ev_ctx_rm`, so we need to pull it out of
- # the DB.
- events_to_check.append(event_id)
- # Check if any of the changes that we don't have events for are joins.
- if events_to_check:
- members = await self.main_store.get_membership_from_event_ids(
- events_to_check
- )
- is_still_joined = any(
- member and member.membership == Membership.JOIN
- for member in members.values()
- )
- if is_still_joined:
- return True
- # None of the new state events are local joins, so we check the database
- # to see if there are any other local users in the room. We ignore users
- # whose state has changed as we've already their new state above.
- users_to_ignore = [
- state_key
- for typ, state_key in itertools.chain(delta.to_insert, delta.to_delete)
- if typ == EventTypes.Member and self.is_mine_id(state_key)
- ]
- if await self.main_store.is_local_host_in_room_ignoring_users(
- room_id, users_to_ignore
- ):
- return True
- # The server will leave the room, so we go and find out which remote
- # users will still be joined when we leave.
- if current_state is None:
- current_state = await self.main_store.get_partial_current_state_ids(room_id)
- current_state = dict(current_state)
- for key in delta.to_delete:
- current_state.pop(key, None)
- current_state.update(delta.to_insert)
- remote_event_ids = [
- event_id
- for (
- typ,
- state_key,
- ), event_id in current_state.items()
- if typ == EventTypes.Member and not self.is_mine_id(state_key)
- ]
- members = await self.main_store.get_membership_from_event_ids(remote_event_ids)
- potentially_left_users.update(
- member.user_id
- for member in members.values()
- if member and member.membership == Membership.JOIN
- )
- return False
- async def _handle_potentially_left_users(self, user_ids: Set[str]) -> None:
- """Given a set of remote users check if the server still shares a room with
- them. If not then mark those users' device cache as stale.
- """
- if not user_ids:
- return
- joined_users = await self.main_store.get_users_server_still_shares_room_with(
- user_ids
- )
- left_users = user_ids - joined_users
- for user_id in left_users:
- await self.main_store.mark_remote_user_device_list_as_unsubscribed(user_id)
|