From 87e4907755c7ed89dac890052d6712bb2c4205f3 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Wed, 13 Apr 2022 13:05:59 +0100 Subject: [PATCH 1/7] Correctly invalidate `is_partial_state_event` cache Forgot this in #12394. --- synapse/storage/databases/main/state.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/storage/databases/main/state.py b/synapse/storage/databases/main/state.py index eba35f37000d..e6340f8285f4 100644 --- a/synapse/storage/databases/main/state.py +++ b/synapse/storage/databases/main/state.py @@ -396,6 +396,7 @@ def _update_state_for_partial_state_event_txn( ) # TODO(faster_joins): need to do something about workers here + txn.call_after(self.is_partial_state_event.invalidate, (event.event_id,)) txn.call_after( self._get_state_group_for_event.prefill, (event.event_id,), From 5e0ebc0205ba7759f2a7f0450c9c79fed0ead3be Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Wed, 6 Apr 2022 14:42:09 +0100 Subject: [PATCH 2/7] Add a mechanism to await un-partial-stating When we join a room via the faster-joins mechanism, we end up with "partial state" at some points on the event DAG. Many parts of the codebase need to wait for the full state to load. So, we implement a mechanism to keep track of which events have partial state, and wait for them to be fully-populated. --- changelog.d/12399.misc | 1 + synapse/handlers/federation_event.py | 1 + .../storage/databases/main/events_worker.py | 10 +- synapse/storage/state.py | 5 + .../util/partial_state_events_tracker.py | 116 ++++++++++++++++++ tests/storage/util/__init__.py | 13 ++ .../util/test_partial_state_events_tracker.py | 94 ++++++++++++++ 7 files changed, 239 insertions(+), 1 deletion(-) create mode 100644 changelog.d/12399.misc create mode 100644 synapse/storage/util/partial_state_events_tracker.py create mode 100644 tests/storage/util/__init__.py create mode 100644 tests/storage/util/test_partial_state_events_tracker.py diff --git a/changelog.d/12399.misc b/changelog.d/12399.misc new file mode 100644 index 000000000000..cd2e09626d35 --- /dev/null +++ b/changelog.d/12399.misc @@ -0,0 +1 @@ +Preparation for faster-room-join work: Implement a tracking mechanism to allow functions to wait for full room state to arrive. diff --git a/synapse/handlers/federation_event.py b/synapse/handlers/federation_event.py index 32bf02818c54..693b544286ec 100644 --- a/synapse/handlers/federation_event.py +++ b/synapse/handlers/federation_event.py @@ -515,6 +515,7 @@ async def update_state_for_partial_state_event( ) return await self._store.update_state_for_partial_state_event(event, context) + self._state_store.notify_event_un_partial_stated(event.event_id) async def backfill( self, dest: str, room_id: str, limit: int, extremities: Collection[str] diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py index 5288cdba035d..d7fad88027ea 100644 --- a/synapse/storage/databases/main/events_worker.py +++ b/synapse/storage/databases/main/events_worker.py @@ -1956,7 +1956,15 @@ def get_event_id_for_timestamp_txn(txn: LoggingTransaction) -> Optional[str]: async def get_partial_state_events( self, event_ids: Collection[str] ) -> Dict[str, bool]: - """Checks which of the given events have partial state""" + """Checks which of the given events have partial state + + Args: + event_ids: the events we want to check for partial state. + + Returns: + a dict mapping from event id to partial-stateness. We return True for + any of the events which are unknown (or are outliers). + """ result = await self.db_pool.simple_select_many_batch( table="partial_state_events", column="event_id", diff --git a/synapse/storage/state.py b/synapse/storage/state.py index cda194e8c8b7..6537f286f67a 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -31,6 +31,7 @@ from synapse.api.constants import EventTypes from synapse.events import EventBase +from synapse.storage.util.partial_state_events_tracker import PartialStateEventsTracker from synapse.types import MutableStateMap, StateKey, StateMap if TYPE_CHECKING: @@ -542,6 +543,10 @@ class StateGroupStorage: def __init__(self, hs: "HomeServer", stores: "Databases"): self.stores = stores + self._partial_state_events_tracker = PartialStateEventsTracker(stores.main) + + def notify_event_un_partial_stated(self, event_id: str) -> None: + self._partial_state_events_tracker.notify_un_partial_stated(event_id) async def get_state_group_delta( self, state_group: int diff --git a/synapse/storage/util/partial_state_events_tracker.py b/synapse/storage/util/partial_state_events_tracker.py new file mode 100644 index 000000000000..c537cd0831db --- /dev/null +++ b/synapse/storage/util/partial_state_events_tracker.py @@ -0,0 +1,116 @@ +# Copyright 2022 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 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.storage.databases.main.events_worker import EventsWorkerStore +from synapse.util import unwrapFirstError + +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 + self._observers: Dict[str, Set[Deferred]] = 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) + + 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(): + if not partial: + 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] diff --git a/tests/storage/util/__init__.py b/tests/storage/util/__init__.py new file mode 100644 index 000000000000..3a5f22c02235 --- /dev/null +++ b/tests/storage/util/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2022 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. diff --git a/tests/storage/util/test_partial_state_events_tracker.py b/tests/storage/util/test_partial_state_events_tracker.py new file mode 100644 index 000000000000..adab8902fa7e --- /dev/null +++ b/tests/storage/util/test_partial_state_events_tracker.py @@ -0,0 +1,94 @@ +# Copyright 2022 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. + +from typing import Dict +from unittest import mock + +from twisted.internet.defer import CancelledError, ensureDeferred + +from synapse.storage.util.partial_state_events_tracker import PartialStateEventsTracker + +from tests.unittest import TestCase + + +class PartialStateEventsTrackerTestCase(TestCase): + def setUp(self) -> None: + # the results to be returned by the mocked get_partial_state_events + self._events_dict: Dict[str, bool] = {} + + async def get_partial_state_events(events): + return {e: self._events_dict[e] for e in events} + + self.mock_store = mock.Mock(spec_set=["get_partial_state_events"]) + self.mock_store.get_partial_state_events.side_effect = get_partial_state_events + + self.tracker = PartialStateEventsTracker(self.mock_store) + + def test_does_not_block_for_full_state_events(self): + self._events_dict = {"event1": False, "event2": False} + + self.successResultOf( + ensureDeferred(self.tracker.await_full_state(["event1", "event2"])) + ) + + self.mock_store.get_partial_state_events.assert_called_once_with( + ["event1", "event2"] + ) + + def test_blocks_for_partial_state_events(self): + self._events_dict = {"event1": True, "event2": False} + + d = ensureDeferred(self.tracker.await_full_state(["event1", "event2"])) + + # there should be no result yet + self.assertNoResult(d) + + # notifying that the event has been de-partial-stated should unblock + self.tracker.notify_un_partial_stated("event1") + self.successResultOf(d) + + def test_un_partial_state_race(self): + # if the event is un-partial-stated between the initial check and the + # registration of the listener, it should not block. + self._events_dict = {"event1": True, "event2": False} + + async def get_partial_state_events(events): + res = {e: self._events_dict[e] for e in events} + # change the result for next time + self._events_dict = {"event1": False, "event2": False} + return res + + self.mock_store.get_partial_state_events.side_effect = get_partial_state_events + + self.successResultOf( + ensureDeferred(self.tracker.await_full_state(["event1", "event2"])) + ) + + def test_cancellation(self): + self._events_dict = {"event1": True, "event2": False} + + d1 = ensureDeferred(self.tracker.await_full_state(["event1", "event2"])) + self.assertNoResult(d1) + + d2 = ensureDeferred(self.tracker.await_full_state(["event1"])) + self.assertNoResult(d2) + + d1.cancel() + self.assertFailure(d1, CancelledError) + + # d2 should still be waiting! + self.assertNoResult(d2) + + self.tracker.notify_un_partial_stated("event1") + self.successResultOf(d2) From df5a3c4ffd05e053847c3e19fba78a12c373a092 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Wed, 13 Apr 2022 10:27:37 +0100 Subject: [PATCH 3/7] `StateGroupStorage`: wait for full state before returning results --- synapse/storage/state.py | 23 ++++++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/synapse/storage/state.py b/synapse/storage/state.py index 6537f286f67a..d1d58592145e 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -584,7 +584,7 @@ async def get_state_groups_ids( if not event_ids: return {} - event_to_groups = await self.stores.main._get_state_group_for_events(event_ids) + event_to_groups = await self._get_state_group_for_events(event_ids) groups = set(event_to_groups.values()) group_to_state = await self.stores.state._get_state_for_groups(groups) @@ -673,7 +673,7 @@ async def get_state_for_events( RuntimeError if we don't have a state group for one or more of the events (ie they are outliers or unknown) """ - event_to_groups = await self.stores.main._get_state_group_for_events(event_ids) + event_to_groups = await self._get_state_group_for_events(event_ids) groups = set(event_to_groups.values()) group_to_state = await self.stores.state._get_state_for_groups( @@ -714,7 +714,7 @@ async def get_state_ids_for_events( RuntimeError if we don't have a state group for one or more of the events (ie they are outliers or unknown) """ - event_to_groups = await self.stores.main._get_state_group_for_events(event_ids) + event_to_groups = await self._get_state_group_for_events(event_ids) groups = set(event_to_groups.values()) group_to_state = await self.stores.state._get_state_for_groups( @@ -790,6 +790,23 @@ def _get_state_for_groups( groups, state_filter or StateFilter.all() ) + async def _get_state_group_for_events( + self, + event_ids: Collection[str], + await_full_state: bool = True, + ) -> Mapping[str, int]: + """Returns mapping event_id -> state_group + + Args: + event_ids: events to get state groups for + await_full_state: if true, will block if we do not yet have complete + state at this event. + """ + if await_full_state: + await self._partial_state_events_tracker.await_full_state(event_ids) + + return await self.stores.main._get_state_group_for_events(event_ids) + async def store_state_group( self, event_id: str, From 13c86914f2cfc2e6315fd70802ce663737deada4 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Fri, 18 Feb 2022 18:50:57 +0000 Subject: [PATCH 4/7] Enable complement tests --- docker/complement/conf/homeserver.yaml | 4 +++- scripts-dev/complement.sh | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/docker/complement/conf/homeserver.yaml b/docker/complement/conf/homeserver.yaml index c9d6a312401a..174f87f52ee0 100644 --- a/docker/complement/conf/homeserver.yaml +++ b/docker/complement/conf/homeserver.yaml @@ -103,8 +103,10 @@ experimental_features: spaces_enabled: true # Enable history backfilling support msc2716_enabled: true - # server-side support for partial state in /send_join + # server-side support for partial state in /send_join responses msc3706_enabled: true + # client-side support for partial state in /send_join responses + faster_joins: true # Enable jump to date endpoint msc3030_enabled: true diff --git a/scripts-dev/complement.sh b/scripts-dev/complement.sh index d34e9f355483..e0feba05fa4f 100755 --- a/scripts-dev/complement.sh +++ b/scripts-dev/complement.sh @@ -64,4 +64,4 @@ docker build -t $COMPLEMENT_BASE_IMAGE -f "docker/complement/$COMPLEMENT_DOCKERF # Run the tests! echo "Images built; running complement" cd "$COMPLEMENT_DIR" -go test -v -tags synapse_blacklist,msc2716,msc3030 -count=1 "$@" ./tests/... +go test -v -tags synapse_blacklist,msc2716,msc3030,faster_joins -count=1 "$@" ./tests/... From d496e9599dad1cd609508d3c0e0a04c198f7863e Mon Sep 17 00:00:00 2001 From: Richard van der Hoff <1389908+richvdh@users.noreply.github.com> Date: Wed, 20 Apr 2022 16:18:05 +0100 Subject: [PATCH 5/7] Update synapse/storage/util/partial_state_events_tracker.py Co-authored-by: Sean Quah <8349537+squahtx@users.noreply.github.com> --- synapse/storage/util/partial_state_events_tracker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/storage/util/partial_state_events_tracker.py b/synapse/storage/util/partial_state_events_tracker.py index c537cd0831db..6a02f81eb3a8 100644 --- a/synapse/storage/util/partial_state_events_tracker.py +++ b/synapse/storage/util/partial_state_events_tracker.py @@ -31,7 +31,7 @@ class PartialStateEventsTracker: def __init__(self, store: EventsWorkerStore): self._store = store - self._observers: Dict[str, Set[Deferred]] = defaultdict(set) + 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 From 3765f7ae15675c555efd74d9c90f055509ebe36e Mon Sep 17 00:00:00 2001 From: Richard van der Hoff <1389908+richvdh@users.noreply.github.com> Date: Wed, 20 Apr 2022 16:19:11 +0100 Subject: [PATCH 6/7] Update synapse/storage/util/partial_state_events_tracker.py --- synapse/storage/util/partial_state_events_tracker.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/synapse/storage/util/partial_state_events_tracker.py b/synapse/storage/util/partial_state_events_tracker.py index 6a02f81eb3a8..0c03a49633c9 100644 --- a/synapse/storage/util/partial_state_events_tracker.py +++ b/synapse/storage/util/partial_state_events_tracker.py @@ -31,6 +31,8 @@ class PartialStateEventsTracker: 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: From 87398730eb6d258ba1666a9e27e55653eede644c Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Wed, 20 Apr 2022 16:30:33 +0100 Subject: [PATCH 7/7] Fix race causing `AlreadyCalledError` --- .../util/partial_state_events_tracker.py | 4 +++- .../util/test_partial_state_events_tracker.py | 23 +++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/synapse/storage/util/partial_state_events_tracker.py b/synapse/storage/util/partial_state_events_tracker.py index 0c03a49633c9..a61a951ef0c5 100644 --- a/synapse/storage/util/partial_state_events_tracker.py +++ b/synapse/storage/util/partial_state_events_tracker.py @@ -94,7 +94,9 @@ async def await_full_state(self, event_ids: Collection[str]) -> None: for event_id, partial in ( await self._store.get_partial_state_events(observers.keys()) ).items(): - if not partial: + # 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( diff --git a/tests/storage/util/test_partial_state_events_tracker.py b/tests/storage/util/test_partial_state_events_tracker.py index adab8902fa7e..303e190b6cc2 100644 --- a/tests/storage/util/test_partial_state_events_tracker.py +++ b/tests/storage/util/test_partial_state_events_tracker.py @@ -75,6 +75,29 @@ async def get_partial_state_events(events): ensureDeferred(self.tracker.await_full_state(["event1", "event2"])) ) + def test_un_partial_state_during_get_partial_state_events(self): + # we should correctly handle a call to notify_un_partial_stated during the + # second call to get_partial_state_events. + + self._events_dict = {"event1": True, "event2": False} + + async def get_partial_state_events1(events): + self.mock_store.get_partial_state_events.side_effect = ( + get_partial_state_events2 + ) + return {e: self._events_dict[e] for e in events} + + async def get_partial_state_events2(events): + self.tracker.notify_un_partial_stated("event1") + self._events_dict["event1"] = False + return {e: self._events_dict[e] for e in events} + + self.mock_store.get_partial_state_events.side_effect = get_partial_state_events1 + + self.successResultOf( + ensureDeferred(self.tracker.await_full_state(["event1", "event2"])) + ) + def test_cancellation(self): self._events_dict = {"event1": True, "event2": False}