mirror of
https://mau.dev/maunium/synapse.git
synced 2024-09-05 17:12:37 +00:00
![Erik Johnston](/assets/img/avatar_default.png)
During the migration the automated script to update the copyright headers accidentally got rid of some of the existing copyright lines. Reinstate them.
195 lines
7.2 KiB
Python
195 lines
7.2 KiB
Python
#
|
|
# This file is licensed under the Affero General Public License (AGPL) version 3.
|
|
#
|
|
# Copyright 2022 The Matrix.org Foundation C.I.C.
|
|
# Copyright (C) 2023 New Vector, Ltd
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# See the GNU Affero General Public License for more details:
|
|
# <https://www.gnu.org/licenses/agpl-3.0.html>.
|
|
#
|
|
# Originally licensed under the Apache License, Version 2.0:
|
|
# <http://www.apache.org/licenses/LICENSE-2.0>.
|
|
#
|
|
# [This file includes modifications made by New Vector Limited]
|
|
#
|
|
#
|
|
|
|
import logging
|
|
from collections import defaultdict
|
|
from typing import Collection, Dict, Set
|
|
|
|
from twisted.internet import defer
|
|
from twisted.internet.defer import Deferred
|
|
|
|
from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable
|
|
from synapse.logging.opentracing import trace_with_opname
|
|
from synapse.storage.databases.main.events_worker import EventsWorkerStore
|
|
from synapse.storage.databases.main.room import RoomWorkerStore
|
|
from synapse.util import unwrapFirstError
|
|
from synapse.util.cancellation import cancellable
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PartialStateEventsTracker:
|
|
"""Keeps track of which events have partial state, after a partial-state join"""
|
|
|
|
def __init__(self, store: EventsWorkerStore):
|
|
self._store = store
|
|
# a map from event id to a set of Deferreds which are waiting for that event to be
|
|
# un-partial-stated.
|
|
self._observers: Dict[str, Set[Deferred[None]]] = defaultdict(set)
|
|
|
|
def notify_un_partial_stated(self, event_id: str) -> None:
|
|
"""Notify that we now have full state for a given event
|
|
|
|
Called by the state-resynchronization loop whenever we resynchronize the state
|
|
for a particular event. Unblocks any callers to await_full_state() for that
|
|
event.
|
|
|
|
Args:
|
|
event_id: the event that now has full state.
|
|
"""
|
|
observers = self._observers.pop(event_id, None)
|
|
if not observers:
|
|
return
|
|
logger.info(
|
|
"Notifying %i things waiting for un-partial-stating of event %s",
|
|
len(observers),
|
|
event_id,
|
|
)
|
|
with PreserveLoggingContext():
|
|
for o in observers:
|
|
o.callback(None)
|
|
|
|
@trace_with_opname("PartialStateEventsTracker.await_full_state")
|
|
@cancellable
|
|
async def await_full_state(self, event_ids: Collection[str]) -> None:
|
|
"""Wait for all the given events to have full state.
|
|
|
|
Args:
|
|
event_ids: the list of event ids that we want full state for
|
|
"""
|
|
# first try the happy path: if there are no partial-state events, we can return
|
|
# quickly
|
|
partial_state_event_ids = [
|
|
ev
|
|
for ev, p in (await self._store.get_partial_state_events(event_ids)).items()
|
|
if p
|
|
]
|
|
|
|
if not partial_state_event_ids:
|
|
return
|
|
|
|
logger.info(
|
|
"Awaiting un-partial-stating of events %s",
|
|
partial_state_event_ids,
|
|
stack_info=True,
|
|
)
|
|
|
|
# create an observer for each lazy-joined event
|
|
observers: Dict[str, Deferred[None]] = {
|
|
event_id: Deferred() for event_id in partial_state_event_ids
|
|
}
|
|
for event_id, observer in observers.items():
|
|
self._observers[event_id].add(observer)
|
|
|
|
try:
|
|
# some of them may have been un-lazy-joined between us checking the db and
|
|
# registering the observer, in which case we'd wait forever for the
|
|
# notification. Call back the observers now.
|
|
for event_id, partial in (
|
|
await self._store.get_partial_state_events(observers.keys())
|
|
).items():
|
|
# there may have been a call to notify_un_partial_stated during the
|
|
# db query, so the observers may already have been called.
|
|
if not partial and not observers[event_id].called:
|
|
observers[event_id].callback(None)
|
|
|
|
await make_deferred_yieldable(
|
|
defer.gatherResults(
|
|
observers.values(),
|
|
consumeErrors=True,
|
|
)
|
|
).addErrback(unwrapFirstError)
|
|
logger.info("Events %s all un-partial-stated", observers.keys())
|
|
finally:
|
|
# remove any observers we created. This should happen when the notification
|
|
# is received, but that might not happen for two reasons:
|
|
# (a) we're bailing out early on an exception (including us being
|
|
# cancelled during the await)
|
|
# (b) the event got de-lazy-joined before we set up the observer.
|
|
for event_id, observer in observers.items():
|
|
observer_set = self._observers.get(event_id)
|
|
if observer_set:
|
|
observer_set.discard(observer)
|
|
if not observer_set:
|
|
del self._observers[event_id]
|
|
|
|
|
|
class PartialCurrentStateTracker:
|
|
"""Keeps track of which rooms have partial state, after partial-state joins"""
|
|
|
|
def __init__(self, store: RoomWorkerStore):
|
|
self._store = store
|
|
|
|
# a map from room id to a set of Deferreds which are waiting for that room to be
|
|
# un-partial-stated.
|
|
self._observers: Dict[str, Set[Deferred[None]]] = defaultdict(set)
|
|
|
|
def notify_un_partial_stated(self, room_id: str) -> None:
|
|
"""Notify that we now have full current state for a given room
|
|
|
|
Unblocks any callers to await_full_state() for that room.
|
|
|
|
Args:
|
|
room_id: the room that now has full current state.
|
|
"""
|
|
observers = self._observers.pop(room_id, None)
|
|
if not observers:
|
|
return
|
|
logger.info(
|
|
"Notifying %i things waiting for un-partial-stating of room %s",
|
|
len(observers),
|
|
room_id,
|
|
)
|
|
with PreserveLoggingContext():
|
|
for o in observers:
|
|
o.callback(None)
|
|
|
|
@trace_with_opname("PartialCurrentStateTracker.await_full_state")
|
|
@cancellable
|
|
async def await_full_state(self, room_id: str) -> None:
|
|
# We add the deferred immediately so that the DB call to check for
|
|
# partial state doesn't race when we unpartial the room.
|
|
d: Deferred[None] = Deferred()
|
|
self._observers.setdefault(room_id, set()).add(d)
|
|
|
|
try:
|
|
# Check if the room has partial current state or not.
|
|
has_partial_state = await self._store.is_partial_state_room(room_id)
|
|
if not has_partial_state:
|
|
return
|
|
|
|
logger.info(
|
|
"Awaiting un-partial-stating of room %s",
|
|
room_id,
|
|
stack_info=True,
|
|
)
|
|
|
|
await make_deferred_yieldable(d)
|
|
|
|
logger.info("Room has un-partial-stated")
|
|
finally:
|
|
# Remove the added observer, and remove the room entry if its empty.
|
|
ds = self._observers.get(room_id)
|
|
if ds is not None:
|
|
ds.discard(d)
|
|
if not ds:
|
|
self._observers.pop(room_id, None)
|