Add changelog entry

.github/workflows/fix_lint.yaml

@@ -21,7 +21,7 @@ jobs:
           # We use nightly so that `fmt` correctly groups together imports, and
           # clippy correctly fixes up the benchmarks.
           toolchain: nightly-2022-12-01
-          components: clippy, rustfmt
+          components: rustfmt
       - uses: Swatinem/rust-cache@v2
 
       - name: Setup Poetry

CHANGES.md

@@ -1,39 +1,3 @@
-# Synapse 1.124.0rc2 (2025-02-05)
-
-### Bugfixes
-
-- Fix regression where persisting events in some rooms could fail after a previous unclean shutdown. Introduced in v1.124.0rc1. ([\#18137](https://github.com/element-hq/synapse/issues/18137))
-
-
-
-
-# Synapse 1.124.0rc1 (2025-02-04)
-
-### Bugfixes
-
-- Add rate limit `rc_presence.per_user`. This prevents load from excessive presence updates sent by clients via sync api. Also rate limit `/_matrix/client/v3/presence` as per the spec. Contributed by @rda0. ([\#18000](https://github.com/element-hq/synapse/issues/18000))
-- Deactivated users will no longer automatically accept an invite when `auto_accept_invites` is enabled. ([\#18073](https://github.com/element-hq/synapse/issues/18073))
-- Fix join being denied after being invited over federation. Also fixes other out-of-band membership transitions. ([\#18075](https://github.com/element-hq/synapse/issues/18075))
-- Updates contributed `docker-compose.yml` file to PostgreSQL v15, as v12 is no longer supported by Synapse.
-  Contributed by @maxkratz. ([\#18089](https://github.com/element-hq/synapse/issues/18089))
-- Fix rare edge case where state groups could be deleted while we are persisting new events that reference them. ([\#18107](https://github.com/element-hq/synapse/issues/18107), [\#18130](https://github.com/element-hq/synapse/issues/18130), [\#18131](https://github.com/element-hq/synapse/issues/18131))
-- Raise an error if someone is using an incorrect suffix in a config duration string. ([\#18112](https://github.com/element-hq/synapse/issues/18112))
-- Fix a bug where the [Delete Room Admin API](https://element-hq.github.io/synapse/latest/admin_api/rooms.html#version-2-new-version) would fail if the `block` parameter was set to `true` and a worker other than the main process was configured to handle background tasks. ([\#18119](https://github.com/element-hq/synapse/issues/18119))
-
-### Internal Changes
-
-- Increase the length of the generated `nonce` parameter when perfoming OIDC logins to comply with the TI-Messenger spec. ([\#18109](https://github.com/element-hq/synapse/issues/18109))
-
-
-
-### Updates to locked dependencies
-
-* Bump dawidd6/action-download-artifact from 7 to 8. ([\#18108](https://github.com/element-hq/synapse/issues/18108))
-* Bump log from 0.4.22 to 0.4.25. ([\#18098](https://github.com/element-hq/synapse/issues/18098))
-* Bump python-multipart from 0.0.18 to 0.0.20. ([\#18096](https://github.com/element-hq/synapse/issues/18096))
-* Bump serde_json from 1.0.135 to 1.0.137. ([\#18099](https://github.com/element-hq/synapse/issues/18099))
-* Bump types-bleach from 6.1.0.20240331 to 6.2.0.20241123. ([\#18082](https://github.com/element-hq/synapse/issues/18082))
-
 # Synapse 1.123.0 (2025-01-28)
 
 No significant changes since 1.123.0rc1.

LICENSE-COMMERCIAL

@@ -1,6 +0,0 @@
-Licensees holding a valid commercial license with Element may use this
-software in accordance with the terms contained in a written agreement
-between you and Element.
-
-To purchase a commercial license please contact our sales team at
-licensing@element.io

README.rst

@@ -10,15 +10,14 @@ implementation, written and maintained by `Element <https://element.io>`_.
 `Matrix <https://github.com/matrix-org>`__ is the open standard for
 secure and interoperable real time communications. You can directly run
 and manage the source code in this repository, available under an AGPL
-license (or alternatively under a commercial license from Element).
-There is no support provided by Element unless you have a
-subscription from Element.
+license. There is no support provided from Element unless you have a
+subscription.
 
-Subscription
-============
+Subscription alternative
+========================
 
-For those that need an enterprise-ready solution, Element
-Server Suite (ESS) is `available via subscription <https://element.io/pricing>`_.
+Alternatively, for those that need an enterprise-ready solution, Element
+Server Suite (ESS) is `available as a subscription <https://element.io/pricing>`_.
 ESS builds on Synapse to offer a complete Matrix-based backend including the full
 `Admin Console product <https://element.io/enterprise-functionality/admin-console>`_,
 giving admins the power to easily manage an organization-wide
@@ -250,20 +249,6 @@ Developers might be particularly interested in:
 Alongside all that, join our developer community on Matrix:
 `#synapse-dev:matrix.org <https://matrix.to/#/#synapse-dev:matrix.org>`_, featuring real humans!
 
-Copyright and Licensing
-=======================
-
-Copyright 2014-2017 OpenMarket Ltd
-Copyright 2017 Vector Creations Ltd
-Copyright 2017-2025 New Vector Ltd 
-
-This software is dual-licensed by New Vector Ltd (Element). It can be used either:
-  
-(1) for free 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); OR
-  
-(2) under the terms of a paid-for Element Commercial License agreement between you and Element (the terms of which may vary depending on what you and Element have agreed to).
-Unless required by applicable law or agreed to in writing, software distributed under the Licenses is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the Licenses for the specific language governing permissions and limitations under the Licenses.
-
 
 .. |support| image:: https://img.shields.io/badge/matrix-community%20support-success
   :alt: (get community support in #synapse:matrix.org)

changelog.d/18000.bugfix

@@ -0,0 +1 @@
+Add rate limit `rc_presence.per_user`. This prevents load from excessive presence updates sent by clients via sync api. Also rate limit `/_matrix/client/v3/presence` as per the spec. Contributed by @rda0.

changelog.d/18004.feature

@@ -1 +0,0 @@
-Add experimental config options `admin_token_path` and `client_secret_path` for MSC 3861.

changelog.d/18073.bugfix

@@ -0,0 +1 @@
+Deactivated users will no longer automatically accept an invite when `auto_accept_invites` is enabled.

changelog.d/18075.bugfix

@@ -0,0 +1 @@
+Fix join being denied after being invited over federation. Also fixes other out-of-band membership transitions.

changelog.d/18089.bugfix

@@ -0,0 +1,2 @@
+Updates contributed `docker-compose.yml` file to PostgreSQL v15, as v12 is no longer supported by Synapse.
+Contributed by @maxkratz.

changelog.d/18109.misc

@@ -0,0 +1 @@
+Increase the length of the generated `nonce` parameter when perfoming OIDC logins to comply with the TI-Messenger spec.

changelog.d/18112.bugfix

@@ -0,0 +1 @@
+Raise an error if someone is using an incorrect suffix in a config duration string.

changelog.d/18119.bugfix

@@ -0,0 +1 @@
+Fix a bug where the [Delete Room Admin API](https://element-hq.github.io/synapse/latest/admin_api/rooms.html#version-2-new-version) would fail if the `block` parameter was set to `true` and a worker other than the main process was configured to handle background tasks.

changelog.d/18124.misc

@@ -0,0 +1 @@
+Add log message when worker lock timeouts get large.

changelog.d/18134.misc

@@ -1 +0,0 @@
-Make it explicit that you can buy an AGPL-alternative commercial license from Element.

changelog.d/18135.bugfix

@@ -1 +0,0 @@
-Fix user directory search when using a legacy module with a `check_username_for_spam` callback. Broke in v1.122.0.

changelog.d/18136.misc

@@ -1 +0,0 @@
-Fix the 'Fix linting' GitHub Actions workflow.

changelog.d/18139.misc

@@ -1 +0,0 @@
-Do not log exceptions when clients provide empty `since` token to `/sync` API.

debian/changelog

@@ -1,15 +1,3 @@
-matrix-synapse-py3 (1.124.0~rc2) stable; urgency=medium
-
-  * New Synapse release 1.124.0rc2.
-
- -- Synapse Packaging team <packages@matrix.org>  Wed, 05 Feb 2025 16:35:53 +0000
-
-matrix-synapse-py3 (1.124.0~rc1) stable; urgency=medium
-
-  * New Synapse release 1.124.0rc1.
-
- -- Synapse Packaging team <packages@matrix.org>  Tue, 04 Feb 2025 11:53:05 +0000
-
 matrix-synapse-py3 (1.123.0) stable; urgency=medium
 
   * New Synapse release 1.123.0.

pyproject.toml

@@ -97,7 +97,7 @@ module-name = "synapse.synapse_rust"
 
 [tool.poetry]
 name = "matrix-synapse"
-version = "1.124.0rc2"
+version = "1.123.0"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "AGPL-3.0-or-later"

synapse/api/auth/msc3861_delegated.py

@@ -19,7 +19,7 @@
 #
 #
 import logging
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 from urllib.parse import urlencode
 
 from authlib.oauth2 import ClientAuth
@@ -119,7 +119,7 @@ class MSC3861DelegatedAuth(BaseAuth):
         self._clock = hs.get_clock()
         self._http_client = hs.get_proxied_http_client()
         self._hostname = hs.hostname
-        self._admin_token: Callable[[], Optional[str]] = self._config.admin_token
+        self._admin_token = self._config.admin_token
 
         self._issuer_metadata = RetryOnExceptionCachedCall[OpenIDProviderMetadata](
             self._load_metadata
@@ -133,10 +133,9 @@ class MSC3861DelegatedAuth(BaseAuth):
             )
         else:
             # Else use the client secret
-            client_secret = self._config.client_secret()
-            assert client_secret, "No client_secret provided"
+            assert self._config.client_secret, "No client_secret provided"
             self._client_auth = ClientAuth(
-                self._config.client_id, client_secret, auth_method
+                self._config.client_id, self._config.client_secret, auth_method
             )
 
     async def _load_metadata(self) -> OpenIDProviderMetadata:
@@ -284,7 +283,7 @@ class MSC3861DelegatedAuth(BaseAuth):
             requester = await self.get_user_by_access_token(access_token, allow_expired)
 
         # Do not record requests from MAS using the virtual `__oidc_admin` user.
-        if access_token != self._admin_token():
+        if access_token != self._admin_token:
             await self._record_request(request, requester)
 
         if not allow_guest and requester.is_guest:
@@ -325,8 +324,7 @@ class MSC3861DelegatedAuth(BaseAuth):
         token: str,
         allow_expired: bool = False,
     ) -> Requester:
-        admin_token = self._admin_token()
-        if admin_token is not None and token == admin_token:
+        if self._admin_token is not None and token == self._admin_token:
             # XXX: This is a temporary solution so that the admin API can be called by
             # the OIDC provider. This will be removed once we have OIDC client
             # credentials grant support in matrix-authentication-service.

synapse/config/experimental.py

@@ -20,15 +20,14 @@
 #
 
 import enum
-from functools import cache
-from typing import TYPE_CHECKING, Any, Iterable, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 import attr
 import attr.validators
 
 from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersions
 from synapse.config import ConfigError
-from synapse.config._base import Config, RootConfig, read_file
+from synapse.config._base import Config, RootConfig
 from synapse.types import JsonDict
 
 # Determine whether authlib is installed.
@@ -44,12 +43,6 @@ if TYPE_CHECKING:
     from authlib.jose.rfc7517 import JsonWebKey
 
 
-@cache
-def read_secret_from_file_once(file_path: Any, config_path: Iterable[str]) -> str:
-    """Returns the memoized secret read from file."""
-    return read_file(file_path, config_path).strip()
-
-
 class ClientAuthMethod(enum.Enum):
     """List of supported client auth methods."""
 
@@ -70,40 +63,6 @@ def _parse_jwks(jwks: Optional[JsonDict]) -> Optional["JsonWebKey"]:
     return JsonWebKey.import_key(jwks)
 
 
-def _check_client_secret(
-    instance: "MSC3861", _attribute: attr.Attribute, _value: Optional[str]
-) -> None:
-    if instance._client_secret and instance._client_secret_path:
-        raise ConfigError(
-            (
-                "You have configured both "
-                "`experimental_features.msc3861.client_secret` and "
-                "`experimental_features.msc3861.client_secret_path`. "
-                "These are mutually incompatible."
-            ),
-            ("experimental", "msc3861", "client_secret"),
-        )
-    # Check client secret can be retrieved
-    instance.client_secret()
-
-
-def _check_admin_token(
-    instance: "MSC3861", _attribute: attr.Attribute, _value: Optional[str]
-) -> None:
-    if instance._admin_token and instance._admin_token_path:
-        raise ConfigError(
-            (
-                "You have configured both "
-                "`experimental_features.msc3861.admin_token` and "
-                "`experimental_features.msc3861.admin_token_path`. "
-                "These are mutually incompatible."
-            ),
-            ("experimental", "msc3861", "admin_token"),
-        )
-    # Check client secret can be retrieved
-    instance.admin_token()
-
-
 @attr.s(slots=True, frozen=True)
 class MSC3861:
     """Configuration for MSC3861: Matrix architecture change to delegate authentication via OIDC"""
@@ -138,30 +97,15 @@ class MSC3861:
     )
     """The auth method used when calling the introspection endpoint."""
 
-    _client_secret: Optional[str] = attr.ib(
+    client_secret: Optional[str] = attr.ib(
         default=None,
-        validator=[
-            attr.validators.optional(attr.validators.instance_of(str)),
-            _check_client_secret,
-        ],
+        validator=attr.validators.optional(attr.validators.instance_of(str)),
     )
     """
     The client secret to use when calling the introspection endpoint,
     when using any of the client_secret_* client auth methods.
     """
 
-    _client_secret_path: Optional[str] = attr.ib(
-        default=None,
-        validator=[
-            attr.validators.optional(attr.validators.instance_of(str)),
-            _check_client_secret,
-        ],
-    )
-    """
-    Alternative to `client_secret`: allows the secret to be specified in an
-    external file.
-    """
-
     jwk: Optional["JsonWebKey"] = attr.ib(default=None, converter=_parse_jwks)
     """
     The JWKS to use when calling the introspection endpoint,
@@ -189,7 +133,7 @@ class MSC3861:
                 ClientAuthMethod.CLIENT_SECRET_BASIC,
                 ClientAuthMethod.CLIENT_SECRET_JWT,
             )
-            and self.client_secret() is None
+            and self.client_secret is None
         ):
             raise ConfigError(
                 f"A client secret must be provided when using the {value} client auth method",
@@ -208,48 +152,15 @@ class MSC3861:
     )
     """The URL of the My Account page on the OIDC Provider as per MSC2965."""
 
-    _admin_token: Optional[str] = attr.ib(
+    admin_token: Optional[str] = attr.ib(
         default=None,
-        validator=[
-            attr.validators.optional(attr.validators.instance_of(str)),
-            _check_admin_token,
-        ],
+        validator=attr.validators.optional(attr.validators.instance_of(str)),
     )
     """
     A token that should be considered as an admin token.
     This is used by the OIDC provider, to make admin calls to Synapse.
     """
 
-    _admin_token_path: Optional[str] = attr.ib(
-        default=None,
-        validator=[
-            attr.validators.optional(attr.validators.instance_of(str)),
-            _check_admin_token,
-        ],
-    )
-    """
-    Alternative to `admin_token`: allows the secret to be specified in an
-    external file.
-    """
-
-    def client_secret(self) -> Optional[str]:
-        """Returns the secret given via `client_secret` or `client_secret_path`."""
-        if self._client_secret_path:
-            return read_secret_from_file_once(
-                self._client_secret_path,
-                ("experimental_features", "msc3861", "client_secret_path"),
-            )
-        return self._client_secret
-
-    def admin_token(self) -> Optional[str]:
-        """Returns the admin token given via `admin_token` or `admin_token_path`."""
-        if self._admin_token_path:
-            return read_secret_from_file_once(
-                self._admin_token_path,
-                ("experimental_features", "msc3861", "admin_token_path"),
-            )
-        return self._admin_token
-
     def check_config_conflicts(self, root: RootConfig) -> None:
         """Checks for any configuration conflicts with other parts of Synapse.
 

synapse/handlers/federation_event.py

@@ -151,8 +151,6 @@ class FederationEventHandler:
     def __init__(self, hs: "HomeServer"):
         self._clock = hs.get_clock()
         self._store = hs.get_datastores().main
-        self._state_store = hs.get_datastores().state
-        self._state_deletion_store = hs.get_datastores().state_deletion
         self._storage_controllers = hs.get_storage_controllers()
         self._state_storage_controller = self._storage_controllers.state
 
@@ -582,9 +580,7 @@ class FederationEventHandler:
                         room_version.identifier,
                         state_maps_to_resolve,
                         event_map=None,
-                        state_res_store=StateResolutionStore(
-                            self._store, self._state_deletion_store
-                        ),
+                        state_res_store=StateResolutionStore(self._store),
                     )
                 )
             else:
@@ -1183,9 +1179,7 @@ class FederationEventHandler:
                 room_version,
                 state_maps,
                 event_map={event_id: event},
-                state_res_store=StateResolutionStore(
-                    self._store, self._state_deletion_store
-                ),
+                state_res_store=StateResolutionStore(self._store),
             )
 
         except Exception as e:
@@ -1880,9 +1874,7 @@ class FederationEventHandler:
                     room_version,
                     [local_state_id_map, claimed_auth_events_id_map],
                     event_map=None,
-                    state_res_store=StateResolutionStore(
-                        self._store, self._state_deletion_store
-                    ),
+                    state_res_store=StateResolutionStore(self._store),
                 )
             )
         else:
@@ -2022,9 +2014,7 @@ class FederationEventHandler:
                     room_version,
                     state_sets,
                     event_map=None,
-                    state_res_store=StateResolutionStore(
-                        self._store, self._state_deletion_store
-                    ),
+                    state_res_store=StateResolutionStore(self._store),
                 )
             )
         else:

synapse/handlers/worker_lock.py

@@ -19,6 +19,7 @@
 #
 #
 
+import logging
 import random
 from types import TracebackType
 from typing import (
@@ -269,6 +270,10 @@ class WaitingLock:
     def _get_next_retry_interval(self) -> float:
         next = self._retry_interval
         self._retry_interval = max(5, next * 2)
+        if self._retry_interval > 5 * 2 ^ 7:  # ~10 minutes
+            logging.warning(
+                f"Lock timeout is getting excessive: {self._retry_interval}s. There may be a deadlock."
+            )
         return next * random.uniform(0.9, 1.1)
 
 
@@ -344,4 +349,8 @@ class WaitingMultiLock:
     def _get_next_retry_interval(self) -> float:
         next = self._retry_interval
         self._retry_interval = max(5, next * 2)
+        if self._retry_interval > 5 * 2 ^ 7:  # ~10 minutes
+            logging.warning(
+                f"Lock timeout is getting excessive: {self._retry_interval}s. There may be a deadlock."
+            )
         return next * random.uniform(0.9, 1.1)

synapse/module_api/callbacks/spamchecker_callbacks.py

@@ -19,7 +19,6 @@
 #
 #
 
-import functools
 import inspect
 import logging
 from typing import (
@@ -298,7 +297,6 @@ def load_legacy_spam_checkers(hs: "synapse.server.HomeServer") -> None:
                         "Bad signature for callback check_registration_for_spam",
                     )
 
-            @functools.wraps(wrapped_func)
             def run(*args: Any, **kwargs: Any) -> Awaitable:
                 # Assertion required because mypy can't prove we won't change `f`
                 # back to `None`. See

synapse/state/__init__.py

@@ -59,13 +59,11 @@ from synapse.types.state import StateFilter
 from synapse.util.async_helpers import Linearizer
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.util.metrics import Measure, measure_func
-from synapse.util.stringutils import shortstr
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
     from synapse.storage.controllers import StateStorageController
     from synapse.storage.databases.main import DataStore
-    from synapse.storage.databases.state.deletion import StateDeletionDataStore
 
 logger = logging.getLogger(__name__)
 metrics_logger = logging.getLogger("synapse.state.metrics")
@@ -196,8 +194,6 @@ class StateHandler:
         self._storage_controllers = hs.get_storage_controllers()
         self._events_shard_config = hs.config.worker.events_shard_config
         self._instance_name = hs.get_instance_name()
-        self._state_store = hs.get_datastores().state
-        self._state_deletion_store = hs.get_datastores().state_deletion
 
         self._update_current_state_client = (
             ReplicationUpdateCurrentStateRestServlet.make_client(hs)
@@ -479,10 +475,7 @@ class StateHandler:
     @trace
     @measure_func()
     async def resolve_state_groups_for_events(
-        self,
-        room_id: str,
-        event_ids: StrCollection,
-        await_full_state: bool = True,
+        self, room_id: str, event_ids: StrCollection, await_full_state: bool = True
     ) -> _StateCacheEntry:
         """Given a list of event_ids this method fetches the state at each
         event, resolves conflicts between them and returns them.
@@ -518,17 +511,6 @@ class StateHandler:
             ) = await self._state_storage_controller.get_state_group_delta(
                 state_group_id
             )
-
-            if prev_group:
-                # Ensure that we still have the prev group, and ensure we don't
-                # delete it while we're persisting the event.
-                missing_state_group = await self._state_deletion_store.check_state_groups_and_bump_deletion(
-                    {prev_group}
-                )
-                if missing_state_group:
-                    prev_group = None
-                    delta_ids = None
-
             return _StateCacheEntry(
                 state=None,
                 state_group=state_group_id,
@@ -549,9 +531,7 @@ class StateHandler:
             room_version,
             state_to_resolve,
             None,
-            state_res_store=StateResolutionStore(
-                self.store, self._state_deletion_store
-            ),
+            state_res_store=StateResolutionStore(self.store),
         )
         return result
 
@@ -683,25 +663,7 @@ class StateResolutionHandler:
         async with self.resolve_linearizer.queue(group_names):
             cache = self._state_cache.get(group_names, None)
             if cache:
-                # Check that the returned cache entry doesn't point to deleted
-                # state groups.
-                state_groups_to_check = set()
-                if cache.state_group is not None:
-                    state_groups_to_check.add(cache.state_group)
-
-                if cache.prev_group is not None:
-                    state_groups_to_check.add(cache.prev_group)
-
-                missing_state_groups = await state_res_store.state_deletion_store.check_state_groups_and_bump_deletion(
-                    state_groups_to_check
-                )
-
-                if not missing_state_groups:
-                    return cache
-                else:
-                    # There are missing state groups, so let's remove the stale
-                    # entry and continue as if it was a cache miss.
-                    self._state_cache.pop(group_names, None)
+                return cache
 
             logger.info(
                 "Resolving state for %s with groups %s",
@@ -709,16 +671,6 @@ class StateResolutionHandler:
                 list(group_names),
             )
 
-            # We double check that none of the state groups have been deleted.
-            # They shouldn't be as all these state groups should be referenced.
-            missing_state_groups = await state_res_store.state_deletion_store.check_state_groups_and_bump_deletion(
-                group_names
-            )
-            if missing_state_groups:
-                raise Exception(
-                    f"State groups have been deleted: {shortstr(missing_state_groups)}"
-                )
-
             state_groups_histogram.observe(len(state_groups_ids))
 
             new_state = await self.resolve_events_with_store(
@@ -932,8 +884,7 @@ class StateResolutionStore:
     in well defined way.
     """
 
-    main_store: "DataStore"
-    state_deletion_store: "StateDeletionDataStore"
+    store: "DataStore"
 
     def get_events(
         self, event_ids: StrCollection, allow_rejected: bool = False
@@ -948,7 +899,7 @@ class StateResolutionStore:
             An awaitable which resolves to a dict from event_id to event.
         """
 
-        return self.main_store.get_events(
+        return self.store.get_events(
             event_ids,
             redact_behaviour=EventRedactBehaviour.as_is,
             get_prev_content=False,
@@ -969,4 +920,4 @@ class StateResolutionStore:
             An awaitable that resolves to a set of event IDs.
         """
 
-        return self.main_store.get_auth_chain_difference(room_id, state_sets)
+        return self.store.get_auth_chain_difference(room_id, state_sets)

synapse/storage/controllers/persist_events.py

@@ -332,7 +332,6 @@ class EventsPersistenceStorageController:
         # store for now.
         self.main_store = stores.main
         self.state_store = stores.state
-        self._state_deletion_store = stores.state_deletion
 
         assert stores.persist_events
         self.persist_events_store = stores.persist_events
@@ -550,9 +549,7 @@ class EventsPersistenceStorageController:
             room_version,
             state_maps_by_state_group,
             event_map=None,
-            state_res_store=StateResolutionStore(
-                self.main_store, self._state_deletion_store
-            ),
+            state_res_store=StateResolutionStore(self.main_store),
         )
 
         return await res.get_state(self._state_controller, StateFilter.all())
@@ -638,20 +635,15 @@ class EventsPersistenceStorageController:
                     room_id, [e for e, _ in chunk]
                 )
 
-            # Stop the state groups from being deleted while we're persisting
-            # them.
-            async with self._state_deletion_store.persisting_state_group_references(
-                events_and_contexts
-            ):
-                await self.persist_events_store._persist_events_and_state_updates(
-                    room_id,
-                    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,
-                    new_event_links=new_event_links,
-                )
+            await self.persist_events_store._persist_events_and_state_updates(
+                room_id,
+                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,
+                new_event_links=new_event_links,
+            )
 
         return replaced_events
 
@@ -973,9 +965,7 @@ class EventsPersistenceStorageController:
             room_version,
             state_groups,
             events_map,
-            state_res_store=StateResolutionStore(
-                self.main_store, self._state_deletion_store
-            ),
+            state_res_store=StateResolutionStore(self.main_store),
         )
 
         state_resolutions_during_persistence.inc()

synapse/storage/controllers/purge_events.py

@@ -21,10 +21,9 @@
 
 import itertools
 import logging
-from typing import TYPE_CHECKING, Collection, Mapping, Set
+from typing import TYPE_CHECKING, Set
 
 from synapse.logging.context import nested_logging_context
-from synapse.metrics.background_process_metrics import wrap_as_background_process
 from synapse.storage.databases import Databases
 
 if TYPE_CHECKING:
@@ -39,11 +38,6 @@ class PurgeEventsStorageController:
     def __init__(self, hs: "HomeServer", stores: Databases):
         self.stores = stores
 
-        if hs.config.worker.run_background_tasks:
-            self._delete_state_loop_call = hs.get_clock().looping_call(
-                self._delete_state_groups_loop, 60 * 1000
-            )
-
     async def purge_room(self, room_id: str) -> None:
         """Deletes all record of a room"""
 
@@ -74,15 +68,11 @@ class PurgeEventsStorageController:
             logger.info("[purge] finding state groups that can be deleted")
             sg_to_delete = await self._find_unreferenced_groups(state_groups)
 
-            # Mark these state groups as pending deletion, they will actually
-            # get deleted automatically later.
-            await self.stores.state_deletion.mark_state_groups_as_pending_deletion(
-                sg_to_delete
+            await self.stores.state.purge_unreferenced_state_groups(
+                room_id, sg_to_delete
             )
 
-    async def _find_unreferenced_groups(
-        self, state_groups: Collection[int]
-    ) -> Set[int]:
+    async def _find_unreferenced_groups(self, state_groups: Set[int]) -> Set[int]:
         """Used when purging history to figure out which state groups can be
         deleted.
 
@@ -128,78 +118,6 @@ class PurgeEventsStorageController:
             next_to_search |= prevs
             state_groups_seen |= prevs
 
-            # We also check to see if anything referencing the state groups are
-            # also unreferenced. This helps ensure that we delete unreferenced
-            # state groups, if we don't then we will de-delta them when we
-            # delete the other state groups leading to increased DB usage.
-            next_edges = await self.stores.state.get_next_state_groups(current_search)
-            nexts = set(next_edges.keys())
-            nexts -= state_groups_seen
-            next_to_search |= nexts
-            state_groups_seen |= nexts
-
         to_delete = state_groups_seen - referenced_groups
 
         return to_delete
-
-    @wrap_as_background_process("_delete_state_groups_loop")
-    async def _delete_state_groups_loop(self) -> None:
-        """Background task that deletes any state groups that may be pending
-        deletion."""
-
-        while True:
-            next_to_delete = await self.stores.state_deletion.get_next_state_group_collection_to_delete()
-            if next_to_delete is None:
-                break
-
-            (room_id, groups_to_sequences) = next_to_delete
-            made_progress = await self._delete_state_groups(
-                room_id, groups_to_sequences
-            )
-
-            # If no progress was made in deleting the state groups, then we
-            # break to allow a pause before trying again next time we get
-            # called.
-            if not made_progress:
-                break
-
-    async def _delete_state_groups(
-        self, room_id: str, groups_to_sequences: Mapping[int, int]
-    ) -> bool:
-        """Tries to delete the given state groups.
-
-        Returns:
-            Whether we made progress in deleting the state groups (or marking
-            them as referenced).
-        """
-
-        # We double check if any of the state groups have become referenced.
-        # This shouldn't happen, as any usages should cause the state group to
-        # be removed as pending deletion.
-        referenced_state_groups = await self.stores.main.get_referenced_state_groups(
-            groups_to_sequences
-        )
-
-        if referenced_state_groups:
-            # We mark any state groups that have become referenced as being
-            # used.
-            await self.stores.state_deletion.mark_state_groups_as_used(
-                referenced_state_groups
-            )
-
-            # Update list of state groups to remove referenced ones
-            groups_to_sequences = {
-                state_group: sequence_number
-                for state_group, sequence_number in groups_to_sequences.items()
-                if state_group not in referenced_state_groups
-            }
-
-        if not groups_to_sequences:
-            # We made progress here as long as we marked some state groups as
-            # now referenced.
-            return len(referenced_state_groups) > 0
-
-        return await self.stores.state.purge_unreferenced_state_groups(
-            room_id,
-            groups_to_sequences,
-        )

synapse/storage/databases/__init__.py

@@ -26,7 +26,6 @@ from synapse.storage._base import SQLBaseStore
 from synapse.storage.database import DatabasePool, make_conn
 from synapse.storage.databases.main.events import PersistEventsStore
 from synapse.storage.databases.state import StateGroupDataStore
-from synapse.storage.databases.state.deletion import StateDeletionDataStore
 from synapse.storage.engines import create_engine
 from synapse.storage.prepare_database import prepare_database
 
@@ -50,14 +49,12 @@ class Databases(Generic[DataStoreT]):
         main
         state
         persist_events
-        state_deletion
     """
 
     databases: List[DatabasePool]
     main: "DataStore"  # FIXME: https://github.com/matrix-org/synapse/issues/11165: actually an instance of `main_store_class`
     state: StateGroupDataStore
     persist_events: Optional[PersistEventsStore]
-    state_deletion: StateDeletionDataStore
 
     def __init__(self, main_store_class: Type[DataStoreT], hs: "HomeServer"):
         # Note we pass in the main store class here as workers use a different main
@@ -66,7 +63,6 @@ class Databases(Generic[DataStoreT]):
         self.databases = []
         main: Optional[DataStoreT] = None
         state: Optional[StateGroupDataStore] = None
-        state_deletion: Optional[StateDeletionDataStore] = None
         persist_events: Optional[PersistEventsStore] = None
 
         for database_config in hs.config.database.databases:
@@ -118,8 +114,7 @@ class Databases(Generic[DataStoreT]):
                     if state:
                         raise Exception("'state' data store already configured")
 
-                    state_deletion = StateDeletionDataStore(database, db_conn, hs)
-                    state = StateGroupDataStore(database, db_conn, hs, state_deletion)
+                    state = StateGroupDataStore(database, db_conn, hs)
 
                 db_conn.commit()
 
@@ -140,7 +135,7 @@ class Databases(Generic[DataStoreT]):
         if not main:
             raise Exception("No 'main' database configured")
 
-        if not state or not state_deletion:
+        if not state:
             raise Exception("No 'state' database configured")
 
         # We use local variables here to ensure that the databases do not have
@@ -148,4 +143,3 @@ class Databases(Generic[DataStoreT]):
         self.main = main  # type: ignore[assignment]
         self.state = state
         self.persist_events = persist_events
-        self.state_deletion = state_deletion

synapse/storage/databases/state/deletion.py

@@ -1,521 +0,0 @@
-#
-# This file is licensed under the Affero General Public License (AGPL) version 3.
-#
-# Copyright (C) 2025 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>.
-#
-
-
-import contextlib
-from typing import (
-    TYPE_CHECKING,
-    AbstractSet,
-    AsyncIterator,
-    Collection,
-    Mapping,
-    Optional,
-    Set,
-    Tuple,
-)
-
-from synapse.events import EventBase
-from synapse.events.snapshot import EventContext
-from synapse.storage.database import (
-    DatabasePool,
-    LoggingDatabaseConnection,
-    LoggingTransaction,
-    make_in_list_sql_clause,
-)
-from synapse.storage.engines import PostgresEngine
-from synapse.util.stringutils import shortstr
-
-if TYPE_CHECKING:
-    from synapse.server import HomeServer
-
-
-class StateDeletionDataStore:
-    """Manages deletion of state groups in a safe manner.
-
-    Deleting state groups is challenging as before we actually delete them we
-    need to ensure that there are no in-flight events that refer to the state
-    groups that we want to delete.
-
-    To handle this, we take two approaches. First, before we persist any event
-    we ensure that the state group still exists and mark in the
-    `state_groups_persisting` table that the state group is about to be used.
-    (Note that we have to have the extra table here as state groups and events
-    can be in different databases, and thus we can't check for the existence of
-    state groups in the persist event transaction). Once the event has been
-    persisted, we can remove the row from  `state_groups_persisting`. So long as
-    we check that table before deleting state groups, we can ensure that we
-    never persist events that reference deleted state groups, maintaining
-    database integrity.
-
-    However, we want to avoid throwing exceptions so deep in the process of
-    persisting events. So instead of deleting state groups immediately, we mark
-    them as pending/proposed for deletion and wait for a certain amount of time
-    before performing the deletion. When we come to handle new events that
-    reference state groups, we check if they are pending deletion and bump the
-    time for when they'll be deleted (to give a chance for the event to be
-    persisted, or not).
-
-    When deleting, we need to check that state groups remain unreferenced. There
-    is a race here where we a) fetch state groups that are ready for deletion,
-    b) check they're unreferenced, c) the state group becomes referenced but
-    then gets marked as pending deletion again, d) during the deletion
-    transaction we recheck `state_groups_pending_deletion` table again and see
-    that it exists and so continue with the deletion. To prevent this from
-    happening we add a `sequence_number` column to
-    `state_groups_pending_deletion`, and during deletion we ensure that for a
-    state group we're about to delete that the sequence number doesn't change
-    between steps (a) and (d). So long as we always bump the sequence number
-    whenever an event may become used the race can never happen.
-    """
-
-    # How long to wait before we delete state groups. This should be long enough
-    # for any in-flight events to be persisted. If events take longer to persist
-    # and any of the state groups they reference have been deleted, then the
-    # event will fail to persist (as well as any event in the same batch).
-    DELAY_BEFORE_DELETION_MS = 10 * 60 * 1000
-
-    def __init__(
-        self,
-        database: DatabasePool,
-        db_conn: LoggingDatabaseConnection,
-        hs: "HomeServer",
-    ):
-        self._clock = hs.get_clock()
-        self.db_pool = database
-        self._instance_name = hs.get_instance_name()
-
-        with db_conn.cursor(txn_name="_clear_existing_persising") as txn:
-            self._clear_existing_persising(txn)
-
-    def _clear_existing_persising(self, txn: LoggingTransaction) -> None:
-        """On startup we clear any entries in `state_groups_persisting` that
-        match our instance name, in case of a previous unclean shutdown"""
-
-        self.db_pool.simple_delete_txn(
-            txn,
-            table="state_groups_persisting",
-            keyvalues={"instance_name": self._instance_name},
-        )
-
-    async def check_state_groups_and_bump_deletion(
-        self, state_groups: AbstractSet[int]
-    ) -> Collection[int]:
-        """Checks to make sure that the state groups haven't been deleted, and
-        if they're pending deletion we delay it (allowing time for any event
-        that will use them to finish persisting).
-
-        Returns:
-            The state groups that are missing, if any.
-        """
-
-        return await self.db_pool.runInteraction(
-            "check_state_groups_and_bump_deletion",
-            self._check_state_groups_and_bump_deletion_txn,
-            state_groups,
-        )
-
-    def _check_state_groups_and_bump_deletion_txn(
-        self, txn: LoggingTransaction, state_groups: AbstractSet[int]
-    ) -> Collection[int]:
-        existing_state_groups = self._get_existing_groups_with_lock(txn, state_groups)
-
-        self._bump_deletion_txn(txn, existing_state_groups)
-
-        missing_state_groups = state_groups - existing_state_groups
-        if missing_state_groups:
-            return missing_state_groups
-
-        return ()
-
-    def _bump_deletion_txn(
-        self, txn: LoggingTransaction, state_groups: Collection[int]
-    ) -> None:
-        """Update any pending deletions of the state group that they may now be
-        referenced."""
-
-        if not state_groups:
-            return
-
-        now = self._clock.time_msec()
-        if isinstance(self.db_pool.engine, PostgresEngine):
-            clause, args = make_in_list_sql_clause(
-                self.db_pool.engine, "state_group", state_groups
-            )
-            sql = f"""
-                UPDATE state_groups_pending_deletion
-                SET sequence_number = DEFAULT, insertion_ts = ?
-                WHERE {clause}
-            """
-            args.insert(0, now)
-            txn.execute(sql, args)
-        else:
-            rows = self.db_pool.simple_select_many_txn(
-                txn,
-                table="state_groups_pending_deletion",
-                column="state_group",
-                iterable=state_groups,
-                keyvalues={},
-                retcols=("state_group",),
-            )
-            if not rows:
-                return
-
-            state_groups_to_update = [state_group for (state_group,) in rows]
-
-            self.db_pool.simple_delete_many_txn(
-                txn,
-                table="state_groups_pending_deletion",
-                column="state_group",
-                values=state_groups_to_update,
-                keyvalues={},
-            )
-            self.db_pool.simple_insert_many_txn(
-                txn,
-                table="state_groups_pending_deletion",
-                keys=("state_group", "insertion_ts"),
-                values=[(state_group, now) for state_group in state_groups_to_update],
-            )
-
-    def _get_existing_groups_with_lock(
-        self, txn: LoggingTransaction, state_groups: Collection[int]
-    ) -> AbstractSet[int]:
-        """Return which of the given state groups are in the database, and locks
-        those rows with `KEY SHARE` to ensure they don't get concurrently
-        deleted."""
-        clause, args = make_in_list_sql_clause(self.db_pool.engine, "id", state_groups)
-
-        sql = f"""
-            SELECT id FROM state_groups
-            WHERE {clause}
-        """
-        if isinstance(self.db_pool.engine, PostgresEngine):
-            # On postgres we add a row level lock to the rows to ensure that we
-            # conflict with any concurrent DELETEs. `FOR KEY SHARE` lock will
-            # not conflict with other read
-            sql += """
-            FOR KEY SHARE
-            """
-
-        txn.execute(sql, args)
-        return {state_group for (state_group,) in txn}
-
-    @contextlib.asynccontextmanager
-    async def persisting_state_group_references(
-        self, event_and_contexts: Collection[Tuple[EventBase, EventContext]]
-    ) -> AsyncIterator[None]:
-        """Wraps the persistence of the given events and contexts, ensuring that
-        any state groups referenced still exist and that they don't get deleted
-        during this."""
-
-        referenced_state_groups: Set[int] = set()
-        for event, ctx in event_and_contexts:
-            if ctx.rejected or event.internal_metadata.is_outlier():
-                continue
-
-            assert ctx.state_group is not None
-
-            referenced_state_groups.add(ctx.state_group)
-
-            if ctx.state_group_before_event:
-                referenced_state_groups.add(ctx.state_group_before_event)
-
-        if not referenced_state_groups:
-            # We don't reference any state groups, so nothing to do
-            yield
-            return
-
-        await self.db_pool.runInteraction(
-            "mark_state_groups_as_persisting",
-            self._mark_state_groups_as_persisting_txn,
-            referenced_state_groups,
-        )
-
-        error = True
-        try:
-            yield None
-            error = False
-        finally:
-            await self.db_pool.runInteraction(
-                "finish_persisting",
-                self._finish_persisting_txn,
-                referenced_state_groups,
-                error=error,
-            )
-
-    def _mark_state_groups_as_persisting_txn(
-        self, txn: LoggingTransaction, state_groups: Set[int]
-    ) -> None:
-        """Marks the given state groups as being persisted."""
-
-        existing_state_groups = self._get_existing_groups_with_lock(txn, state_groups)
-        missing_state_groups = state_groups - existing_state_groups
-        if missing_state_groups:
-            raise Exception(
-                f"state groups have been deleted: {shortstr(missing_state_groups)}"
-            )
-
-        self.db_pool.simple_insert_many_txn(
-            txn,
-            table="state_groups_persisting",
-            keys=("state_group", "instance_name"),
-            values=[(state_group, self._instance_name) for state_group in state_groups],
-        )
-
-    def _finish_persisting_txn(
-        self, txn: LoggingTransaction, state_groups: Collection[int], error: bool
-    ) -> None:
-        """Mark the state groups as having finished persistence.
-
-        If `error` is true then we assume the state groups were not persisted,
-        and so we do not clear them from the pending deletion table.
-        """
-        self.db_pool.simple_delete_many_txn(
-            txn,
-            table="state_groups_persisting",
-            column="state_group",
-            values=state_groups,
-            keyvalues={"instance_name": self._instance_name},
-        )
-
-        if error:
-            # The state groups may or may not have been persisted, so we need to
-            # bump the deletion to ensure we recheck if they have become
-            # referenced.
-            self._bump_deletion_txn(txn, state_groups)
-            return
-
-        self.db_pool.simple_delete_many_batch_txn(
-            txn,
-            table="state_groups_pending_deletion",
-            keys=("state_group",),
-            values=[(state_group,) for state_group in state_groups],
-        )
-
-    async def mark_state_groups_as_pending_deletion(
-        self, state_groups: Collection[int]
-    ) -> None:
-        """Mark the given state groups as pending deletion"""
-
-        now = self._clock.time_msec()
-
-        await self.db_pool.simple_upsert_many(
-            table="state_groups_pending_deletion",
-            key_names=("state_group",),
-            key_values=[(state_group,) for state_group in state_groups],
-            value_names=("insertion_ts",),
-            value_values=[(now,) for _ in state_groups],
-            desc="mark_state_groups_as_pending_deletion",
-        )
-
-    async def mark_state_groups_as_used(self, state_groups: Collection[int]) -> None:
-        """Mark the given state groups as now being referenced"""
-
-        await self.db_pool.simple_delete_many(
-            table="state_groups_pending_deletion",
-            column="state_group",
-            iterable=state_groups,
-            keyvalues={},
-            desc="mark_state_groups_as_used",
-        )
-
-    async def get_pending_deletions(
-        self, state_groups: Collection[int]
-    ) -> Mapping[int, int]:
-        """Get which state groups are pending deletion.
-
-        Returns:
-            a mapping from state groups that are pending deletion to their
-            sequence number
-        """
-
-        rows = await self.db_pool.simple_select_many_batch(
-            table="state_groups_pending_deletion",
-            column="state_group",
-            iterable=state_groups,
-            retcols=("state_group", "sequence_number"),
-            keyvalues={},
-            desc="get_pending_deletions",
-        )
-
-        return dict(rows)
-
-    def get_state_groups_ready_for_potential_deletion_txn(
-        self,
-        txn: LoggingTransaction,
-        state_groups_to_sequence_numbers: Mapping[int, int],
-    ) -> Collection[int]:
-        """Given a set of state groups, return which state groups can
-        potentially be deleted.
-
-        The state groups must have been checked to see if they remain
-        unreferenced before calling this function.
-
-        Note: This must be called within the same transaction that the state
-        groups are deleted.
-
-        Args:
-            state_groups_to_sequence_numbers: The state groups, and the sequence
-                numbers from before the state groups were checked to see if they
-                were unreferenced.
-
-        Returns:
-            The subset of state groups that can safely be deleted
-
-        """
-
-        if not state_groups_to_sequence_numbers:
-            return state_groups_to_sequence_numbers
-
-        if isinstance(self.db_pool.engine, PostgresEngine):
-            # On postgres we want to lock the rows FOR UPDATE as early as
-            # possible to help conflicts.
-            clause, args = make_in_list_sql_clause(
-                self.db_pool.engine, "id", state_groups_to_sequence_numbers
-            )
-            sql = f"""
-                SELECT id FROM state_groups
-                WHERE {clause}
-                FOR UPDATE
-            """
-            txn.execute(sql, args)
-
-        # Check the deletion status in the DB of the given state groups
-        clause, args = make_in_list_sql_clause(
-            self.db_pool.engine,
-            column="state_group",
-            iterable=state_groups_to_sequence_numbers,
-        )
-
-        sql = f"""
-            SELECT state_group, insertion_ts, sequence_number FROM (
-                SELECT state_group, insertion_ts, sequence_number FROM state_groups_pending_deletion
-                UNION
-                SELECT state_group, null, null FROM state_groups_persisting
-            ) AS s
-            WHERE {clause}
-        """
-
-        txn.execute(sql, args)
-
-        # The above query will return potentially two rows per state group (one
-        # for each table), so we track which state groups have enough time
-        # elapsed and which are not ready to be persisted.
-        ready_to_be_deleted = set()
-        not_ready_to_be_deleted = set()
-
-        now = self._clock.time_msec()
-        for state_group, insertion_ts, sequence_number in txn:
-            if insertion_ts is None:
-                # A null insertion_ts means that we are currently persisting
-                # events that reference the state group, so we don't delete
-                # them.
-                not_ready_to_be_deleted.add(state_group)
-                continue
-
-            # We know this can't be None if insertion_ts is not None
-            assert sequence_number is not None
-
-            # Check if the sequence number has changed, if it has then it
-            # indicates that the state group may have become referenced since we
-            # checked.
-            if state_groups_to_sequence_numbers[state_group] != sequence_number:
-                not_ready_to_be_deleted.add(state_group)
-                continue
-
-            if now - insertion_ts < self.DELAY_BEFORE_DELETION_MS:
-                # Not enough time has elapsed to allow us to delete.
-                not_ready_to_be_deleted.add(state_group)
-                continue
-
-            ready_to_be_deleted.add(state_group)
-
-        can_be_deleted = ready_to_be_deleted - not_ready_to_be_deleted
-        if not_ready_to_be_deleted:
-            # If there are any state groups that aren't ready to be deleted,
-            # then we also need to remove any state groups that are referenced
-            # by them.
-            clause, args = make_in_list_sql_clause(
-                self.db_pool.engine,
-                column="state_group",
-                iterable=state_groups_to_sequence_numbers,
-            )
-            sql = f"""
-                WITH RECURSIVE ancestors(state_group) AS (
-                    SELECT DISTINCT prev_state_group
-                    FROM state_group_edges WHERE {clause}
-                    UNION
-                    SELECT prev_state_group
-                    FROM state_group_edges
-                    INNER JOIN ancestors USING (state_group)
-                )
-                SELECT state_group FROM ancestors
-            """
-            txn.execute(sql, args)
-
-            can_be_deleted.difference_update(state_group for (state_group,) in txn)
-
-        return can_be_deleted
-
-    async def get_next_state_group_collection_to_delete(
-        self,
-    ) -> Optional[Tuple[str, Mapping[int, int]]]:
-        """Get the next set of state groups to try and delete
-
-        Returns:
-            2-tuple of room_id and mapping of state groups to sequence number.
-        """
-        return await self.db_pool.runInteraction(
-            "get_next_state_group_collection_to_delete",
-            self._get_next_state_group_collection_to_delete_txn,
-        )
-
-    def _get_next_state_group_collection_to_delete_txn(
-        self,
-        txn: LoggingTransaction,
-    ) -> Optional[Tuple[str, Mapping[int, int]]]:
-        """Implementation of `get_next_state_group_collection_to_delete`"""
-
-        # We want to return chunks of state groups that were marked for deletion
-        # at the same time (this isn't necessary, just more efficient). We do
-        # this by looking for the oldest insertion_ts, and then pulling out all
-        # rows that have the same insertion_ts (and room ID).
-        now = self._clock.time_msec()
-
-        sql = """
-            SELECT room_id, insertion_ts
-            FROM state_groups_pending_deletion AS sd
-            INNER JOIN state_groups AS sg ON (id = sd.state_group)
-            LEFT JOIN state_groups_persisting AS sp USING (state_group)
-            WHERE insertion_ts < ? AND sp.state_group IS NULL
-            ORDER BY insertion_ts
-            LIMIT 1
-        """
-        txn.execute(sql, (now - self.DELAY_BEFORE_DELETION_MS,))
-        row = txn.fetchone()
-        if not row:
-            return None
-
-        (room_id, insertion_ts) = row
-
-        sql = """
-            SELECT state_group, sequence_number
-            FROM state_groups_pending_deletion AS sd
-            INNER JOIN state_groups AS sg ON (id = sd.state_group)
-            LEFT JOIN state_groups_persisting AS sp USING (state_group)
-            WHERE room_id = ? AND insertion_ts = ? AND sp.state_group IS NULL
-            ORDER BY insertion_ts
-        """
-        txn.execute(sql, (room_id, insertion_ts))
-
-        return room_id, dict(txn)

synapse/storage/databases/state/store.py

@@ -22,10 +22,10 @@
 import logging
 from typing import (
     TYPE_CHECKING,
+    Collection,
     Dict,
     Iterable,
     List,
-    Mapping,
     Optional,
     Set,
     Tuple,
@@ -36,10 +36,7 @@ import attr
 
 from synapse.api.constants import EventTypes
 from synapse.events import EventBase
-from synapse.events.snapshot import (
-    UnpersistedEventContext,
-    UnpersistedEventContextBase,
-)
+from synapse.events.snapshot import UnpersistedEventContext, UnpersistedEventContextBase
 from synapse.logging.opentracing import tag_args, trace
 from synapse.storage._base import SQLBaseStore
 from synapse.storage.database import (
@@ -58,7 +55,6 @@ from synapse.util.cancellation import cancellable
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
-    from synapse.storage.databases.state.deletion import StateDeletionDataStore
 
 logger = logging.getLogger(__name__)
 
@@ -87,10 +83,8 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
         database: DatabasePool,
         db_conn: LoggingDatabaseConnection,
         hs: "HomeServer",
-        state_deletion_store: "StateDeletionDataStore",
     ):
         super().__init__(database, db_conn, hs)
-        self._state_deletion_store = state_deletion_store
 
         # Originally the state store used a single DictionaryCache to cache the
         # event IDs for the state types in a given state group to avoid hammering
@@ -473,15 +467,14 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
             Returns:
                 A list of state groups
             """
-
-            # We need to check that the prev group isn't about to be deleted
-            is_missing = (
-                self._state_deletion_store._check_state_groups_and_bump_deletion_txn(
-                    txn,
-                    {prev_group},
-                )
+            is_in_db = self.db_pool.simple_select_one_onecol_txn(
+                txn,
+                table="state_groups",
+                keyvalues={"id": prev_group},
+                retcol="id",
+                allow_none=True,
             )
-            if is_missing:
+            if not is_in_db:
                 raise Exception(
                     "Trying to persist state with unpersisted prev_group: %r"
                     % (prev_group,)
@@ -553,7 +546,6 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
                     for key, state_id in context.state_delta_due_to_event.items()
                 ],
             )
-
             return events_and_context
 
         return await self.db_pool.runInteraction(
@@ -609,15 +601,14 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
                 The state group if successfully created, or None if the state
                 needs to be persisted as a full state.
             """
-
-            # We need to check that the prev group isn't about to be deleted
-            is_missing = (
-                self._state_deletion_store._check_state_groups_and_bump_deletion_txn(
-                    txn,
-                    {prev_group},
-                )
+            is_in_db = self.db_pool.simple_select_one_onecol_txn(
+                txn,
+                table="state_groups",
+                keyvalues={"id": prev_group},
+                retcol="id",
+                allow_none=True,
             )
-            if is_missing:
+            if not is_in_db:
                 raise Exception(
                     "Trying to persist state with unpersisted prev_group: %r"
                     % (prev_group,)
@@ -735,10 +726,8 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
         )
 
     async def purge_unreferenced_state_groups(
-        self,
-        room_id: str,
-        state_groups_to_sequence_numbers: Mapping[int, int],
-    ) -> bool:
+        self, room_id: str, state_groups_to_delete: Collection[int]
+    ) -> None:
         """Deletes no longer referenced state groups and de-deltas any state
         groups that reference them.
 
@@ -746,31 +735,21 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
             room_id: The room the state groups belong to (must all be in the
                 same room).
             state_groups_to_delete: Set of all state groups to delete.
-
-        Returns:
-            Whether any state groups were actually deleted.
         """
 
-        return await self.db_pool.runInteraction(
+        await self.db_pool.runInteraction(
             "purge_unreferenced_state_groups",
             self._purge_unreferenced_state_groups,
             room_id,
-            state_groups_to_sequence_numbers,
+            state_groups_to_delete,
         )
 
     def _purge_unreferenced_state_groups(
         self,
         txn: LoggingTransaction,
         room_id: str,
-        state_groups_to_sequence_numbers: Mapping[int, int],
-    ) -> bool:
-        state_groups_to_delete = self._state_deletion_store.get_state_groups_ready_for_potential_deletion_txn(
-            txn, state_groups_to_sequence_numbers
-        )
-
-        if not state_groups_to_delete:
-            return False
-
+        state_groups_to_delete: Collection[int],
+    ) -> None:
         logger.info(
             "[purge] found %i state groups to delete", len(state_groups_to_delete)
         )
@@ -833,8 +812,6 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
             [(sg,) for sg in state_groups_to_delete],
         )
 
-        return True
-
     @trace
     @tag_args
     async def get_previous_state_groups(
@@ -849,35 +826,6 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
             A mapping from state group to previous state group.
         """
 
-        rows = cast(
-            List[Tuple[int, int]],
-            await self.db_pool.simple_select_many_batch(
-                table="state_group_edges",
-                column="state_group",
-                iterable=state_groups,
-                keyvalues={},
-                retcols=("state_group", "prev_state_group"),
-                desc="get_previous_state_groups",
-            ),
-        )
-
-        return dict(rows)
-
-    @trace
-    @tag_args
-    async def get_next_state_groups(
-        self, state_groups: Iterable[int]
-    ) -> Dict[int, int]:
-        """Fetch the groups that have the given state groups as their previous
-        state groups.
-
-        Args:
-            state_groups
-
-        Returns:
-            A mapping from state group to previous state group.
-        """
-
         rows = cast(
             List[Tuple[int, int]],
             await self.db_pool.simple_select_many_batch(
@@ -886,7 +834,7 @@ class StateGroupDataStore(StateBackgroundUpdateStore, SQLBaseStore):
                 iterable=state_groups,
                 keyvalues={},
                 retcols=("state_group", "prev_state_group"),
-                desc="get_next_state_groups",
+                desc="get_previous_state_groups",
             ),
         )
 

synapse/storage/schema/__init__.py

@@ -19,7 +19,7 @@
 #
 #
 
-SCHEMA_VERSION = 89  # remember to update the list below when updating
+SCHEMA_VERSION = 88  # remember to update the list below when updating
 """Represents the expectations made by the codebase about the database schema
 
 This should be incremented whenever the codebase changes its requirements on the
@@ -155,9 +155,6 @@ Changes in SCHEMA_VERSION = 88
       be posted in response to a resettable timeout or an on-demand action.
     - Add background update to fix data integrity issue in the
       `sliding_sync_membership_snapshots` -> `forgotten` column
-
-Changes in SCHEMA_VERSION = 89
-    - Add `state_groups_pending_deletion` and `state_groups_persisting` tables.
 """
 
 

synapse/storage/schema/state/delta/89/01_state_groups_deletion.sql

@@ -1,39 +0,0 @@
---
--- This file is licensed under the Affero General Public License (AGPL) version 3.
---
--- Copyright (C) 2025 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>.
-
--- See the `StateDeletionDataStore` for details of these tables.
-
--- We add state groups to this table when we want to later delete them. The
--- `insertion_ts` column indicates when the state group was proposed for
--- deletion (rather than when it should be deleted).
-CREATE TABLE IF NOT EXISTS state_groups_pending_deletion (
-    sequence_number $%AUTO_INCREMENT_PRIMARY_KEY%$,
-    state_group BIGINT NOT NULL,
-    insertion_ts BIGINT NOT NULL
-);
-
-CREATE UNIQUE INDEX state_groups_pending_deletion_state_group ON state_groups_pending_deletion(state_group);
-CREATE INDEX state_groups_pending_deletion_insertion_ts ON state_groups_pending_deletion(insertion_ts);
-
-
--- Holds the state groups the worker is currently persisting.
---
--- The `sequence_number` column of the `state_groups_pending_deletion` table
--- *must* be updated whenever a state group may have become referenced.
-CREATE TABLE IF NOT EXISTS state_groups_persisting (
-    state_group BIGINT NOT NULL,
-    instance_name TEXT NOT NULL,
-    PRIMARY KEY (state_group, instance_name)
-);
-
-CREATE INDEX state_groups_persisting_instance_name ON state_groups_persisting(instance_name);

synapse/types/__init__.py

@@ -664,12 +664,6 @@ class RoomStreamToken(AbstractMultiWriterStreamToken):
 
     @classmethod
     async def parse(cls, store: "PurgeEventsStore", string: str) -> "RoomStreamToken":
-        # Check it looks like a Synapse token first. We do this so that
-        # we don't log exceptions for obviously incorrect tokens.
-        if not string or string[0] not in ("s", "t", "m"):
-            logger.warning("Invalid token %r", string)
-            raise SynapseError(400, f"Invalid room stream token {string:!r}")
-
         try:
             if string[0] == "s":
                 return cls(topological=None, stream=int(string[1:]))

tests/config/test_load.py

@@ -132,8 +132,6 @@ class ConfigLoadingFileTestCase(ConfigFileTestCase):
             "turn_shared_secret_path: /does/not/exist",
             "registration_shared_secret_path: /does/not/exist",
             "macaroon_secret_key_path: /does/not/exist",
-            "experimental_features:\n  msc3861:\n    client_secret_path: /does/not/exist",
-            "experimental_features:\n  msc3861:\n    admin_token_path: /does/not/exist",
             *["redis:\n  enabled: true\n  password_path: /does/not/exist"]
             * (hiredis is not None),
         ]
@@ -159,14 +157,6 @@ class ConfigLoadingFileTestCase(ConfigFileTestCase):
                 "macaroon_secret_key_path: {}",
                 lambda c: c.key.macaroon_secret_key,
             ),
-            (
-                "experimental_features:\n  msc3861:\n    client_secret_path: {}",
-                lambda c: c.experimental.msc3861.client_secret().encode("utf-8"),
-            ),
-            (
-                "experimental_features:\n  msc3861:\n    admin_token_path: {}",
-                lambda c: c.experimental.msc3861.admin_token().encode("utf-8"),
-            ),
             *[
                 (
                     "redis:\n  enabled: true\n  password_path: {}",

tests/handlers/test_federation_event.py

@@ -807,7 +807,6 @@ class FederationEventHandlerTests(unittest.FederatingHomeserverTestCase):
 
         OTHER_USER = f"@user:{self.OTHER_SERVER_NAME}"
         main_store = self.hs.get_datastores().main
-        state_deletion_store = self.hs.get_datastores().state_deletion
 
         # Create the room.
         kermit_user_id = self.register_user("kermit", "test")
@@ -959,9 +958,7 @@ class FederationEventHandlerTests(unittest.FederatingHomeserverTestCase):
                         bert_member_event.event_id: bert_member_event,
                         rejected_kick_event.event_id: rejected_kick_event,
                     },
-                    state_res_store=StateResolutionStore(
-                        main_store, state_deletion_store
-                    ),
+                    state_res_store=StateResolutionStore(main_store),
                 )
             ),
             [bert_member_event.event_id, rejected_kick_event.event_id],
@@ -1006,9 +1003,7 @@ class FederationEventHandlerTests(unittest.FederatingHomeserverTestCase):
                         rejected_power_levels_event.event_id,
                     ],
                     event_map={},
-                    state_res_store=StateResolutionStore(
-                        main_store, state_deletion_store
-                    ),
+                    state_res_store=StateResolutionStore(main_store),
                     full_conflicted_set=set(),
                 )
             ),

tests/handlers/test_oauth_delegation.py

@@ -795,7 +795,7 @@ class MSC3861OAuthDelegation(HomeserverTestCase):
         req = SynapseRequest(channel, self.site)  # type: ignore[arg-type]
         req.client.host = MAS_IPV4_ADDR
         req.requestHeaders.addRawHeader(
-            "Authorization", f"Bearer {self.auth._admin_token()}"
+            "Authorization", f"Bearer {self.auth._admin_token}"
         )
         req.requestHeaders.addRawHeader("User-Agent", MAS_USER_AGENT)
         req.content = BytesIO(b"")

tests/rest/client/test_rooms.py

@@ -742,7 +742,7 @@ class RoomsCreateTestCase(RoomBase):
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
         self.assertTrue("room_id" in channel.json_body)
         assert channel.resource_usage is not None
-        self.assertEqual(36, channel.resource_usage.db_txn_count)
+        self.assertEqual(34, channel.resource_usage.db_txn_count)
 
     def test_post_room_initial_state(self) -> None:
         # POST with initial_state config key, expect new room id
@@ -755,7 +755,7 @@ class RoomsCreateTestCase(RoomBase):
         self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
         self.assertTrue("room_id" in channel.json_body)
         assert channel.resource_usage is not None
-        self.assertEqual(38, channel.resource_usage.db_txn_count)
+        self.assertEqual(36, channel.resource_usage.db_txn_count)
 
     def test_post_room_visibility_key(self) -> None:
         # POST with visibility config key, expect new room id

tests/rest/client/utils.py

@@ -548,7 +548,7 @@ class RestHelper:
         room_id: str,
         event_type: str,
         body: Dict[str, Any],
-        tok: Optional[str] = None,
+        tok: Optional[str],
         expect_code: int = HTTPStatus.OK,
         state_key: str = "",
     ) -> JsonDict:

tests/storage/test_purge.py

@@ -23,7 +23,6 @@ from twisted.test.proto_helpers import MemoryReactor
 from synapse.api.errors import NotFoundError, SynapseError
 from synapse.rest.client import room
 from synapse.server import HomeServer
-from synapse.types.state import StateFilter
 from synapse.util import Clock
 
 from tests.unittest import HomeserverTestCase
@@ -41,8 +40,6 @@ class PurgeTests(HomeserverTestCase):
         self.room_id = self.helper.create_room_as(self.user_id)
 
         self.store = hs.get_datastores().main
-        self.state_store = hs.get_datastores().state
-        self.state_deletion_store = hs.get_datastores().state_deletion
         self._storage_controllers = self.hs.get_storage_controllers()
 
     def test_purge_history(self) -> None:
@@ -131,142 +128,3 @@ class PurgeTests(HomeserverTestCase):
         self.store._invalidate_local_get_event_cache(create_event.event_id)
         self.get_failure(self.store.get_event(create_event.event_id), NotFoundError)
         self.get_failure(self.store.get_event(first["event_id"]), NotFoundError)
-
-    def test_purge_history_deletes_state_groups(self) -> None:
-        """Test that unreferenced state groups get cleaned up after purge"""
-
-        # Send four state changes to the room.
-        first = self.helper.send_state(
-            self.room_id, event_type="m.foo", body={"test": 1}
-        )
-        second = self.helper.send_state(
-            self.room_id, event_type="m.foo", body={"test": 2}
-        )
-        third = self.helper.send_state(
-            self.room_id, event_type="m.foo", body={"test": 3}
-        )
-        last = self.helper.send_state(
-            self.room_id, event_type="m.foo", body={"test": 4}
-        )
-
-        # Get references to the state groups
-        event_to_groups = self.get_success(
-            self.store._get_state_group_for_events(
-                [
-                    first["event_id"],
-                    second["event_id"],
-                    third["event_id"],
-                    last["event_id"],
-                ]
-            )
-        )
-
-        # Get the topological token
-        token = self.get_success(
-            self.store.get_topological_token_for_event(last["event_id"])
-        )
-        token_str = self.get_success(token.to_string(self.hs.get_datastores().main))
-
-        # Purge everything before this topological token
-        self.get_success(
-            self._storage_controllers.purge_events.purge_history(
-                self.room_id, token_str, True
-            )
-        )
-
-        # Advance so that the background jobs to delete the state groups runs
-        self.reactor.advance(
-            1 + self.state_deletion_store.DELAY_BEFORE_DELETION_MS / 1000
-        )
-
-        # We expect all the state groups associated with events above, except
-        # the last one, should return no state.
-        state_groups = self.get_success(
-            self.state_store._get_state_groups_from_groups(
-                list(event_to_groups.values()), StateFilter.all()
-            )
-        )
-        first_state = state_groups[event_to_groups[first["event_id"]]]
-        second_state = state_groups[event_to_groups[second["event_id"]]]
-        third_state = state_groups[event_to_groups[third["event_id"]]]
-        last_state = state_groups[event_to_groups[last["event_id"]]]
-
-        self.assertEqual(first_state, {})
-        self.assertEqual(second_state, {})
-        self.assertEqual(third_state, {})
-        self.assertNotEqual(last_state, {})
-
-    def test_purge_unreferenced_state_group(self) -> None:
-        """Test that purging a room also gets rid of unreferenced state groups
-        it encounters during the purge.
-
-        This is important, as otherwise these unreferenced state groups get
-        "de-deltaed" during the purge process, consuming lots of disk space.
-        """
-
-        self.helper.send(self.room_id, body="test1")
-        state1 = self.helper.send_state(
-            self.room_id, "org.matrix.test", body={"number": 2}
-        )
-        state2 = self.helper.send_state(
-            self.room_id, "org.matrix.test", body={"number": 3}
-        )
-        self.helper.send(self.room_id, body="test4")
-        last = self.helper.send(self.room_id, body="test5")
-
-        # Create an unreferenced state group that has a prev group of one of the
-        # to-be-purged events.
-        prev_group = self.get_success(
-            self.store._get_state_group_for_event(state1["event_id"])
-        )
-        unreferenced_state_group = self.get_success(
-            self.state_store.store_state_group(
-                event_id=last["event_id"],
-                room_id=self.room_id,
-                prev_group=prev_group,
-                delta_ids={("org.matrix.test", ""): state2["event_id"]},
-                current_state_ids=None,
-            )
-        )
-
-        # Get the topological token
-        token = self.get_success(
-            self.store.get_topological_token_for_event(last["event_id"])
-        )
-        token_str = self.get_success(token.to_string(self.hs.get_datastores().main))
-
-        # Purge everything before this topological token
-        self.get_success(
-            self._storage_controllers.purge_events.purge_history(
-                self.room_id, token_str, True
-            )
-        )
-
-        # Advance so that the background jobs to delete the state groups runs
-        self.reactor.advance(
-            1 + self.state_deletion_store.DELAY_BEFORE_DELETION_MS / 1000
-        )
-
-        # We expect that the unreferenced state group has been deleted.
-        row = self.get_success(
-            self.state_store.db_pool.simple_select_one_onecol(
-                table="state_groups",
-                keyvalues={"id": unreferenced_state_group},
-                retcol="id",
-                allow_none=True,
-                desc="test_purge_unreferenced_state_group",
-            )
-        )
-        self.assertIsNone(row)
-
-        # We expect there to now only be one state group for the room, which is
-        # the state group of the last event (as the only outlier).
-        state_groups = self.get_success(
-            self.state_store.db_pool.simple_select_onecol(
-                table="state_groups",
-                keyvalues={"room_id": self.room_id},
-                retcol="id",
-                desc="test_purge_unreferenced_state_group",
-            )
-        )
-        self.assertEqual(len(state_groups), 1)

tests/storage/test_state_deletion.py

@@ -1,475 +0,0 @@
-#
-# This file is licensed under the Affero General Public License (AGPL) version 3.
-#
-# Copyright (C) 2025 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>.
-#
-
-
-import logging
-
-from twisted.test.proto_helpers import MemoryReactor
-
-from synapse.rest import admin
-from synapse.rest.client import login, room
-from synapse.server import HomeServer
-from synapse.util import Clock
-
-from tests.test_utils.event_injection import create_event
-from tests.unittest import HomeserverTestCase
-
-logger = logging.getLogger(__name__)
-
-
-class StateDeletionStoreTestCase(HomeserverTestCase):
-    """Tests for the StateDeletionStore."""
-
-    servlets = [
-        admin.register_servlets,
-        room.register_servlets,
-        login.register_servlets,
-    ]
-
-    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
-        self.store = hs.get_datastores().main
-        self.state_store = hs.get_datastores().state
-        self.state_deletion_store = hs.get_datastores().state_deletion
-        self.purge_events = hs.get_storage_controllers().purge_events
-
-        # We want to disable the automatic deletion of state groups in the
-        # background, so we can do controlled tests.
-        self.purge_events._delete_state_loop_call.stop()
-
-        self.user_id = self.register_user("test", "password")
-        tok = self.login("test", "password")
-        self.room_id = self.helper.create_room_as(self.user_id, tok=tok)
-
-    def check_if_can_be_deleted(self, state_group: int) -> bool:
-        """Check if the state group is pending deletion."""
-
-        state_group_to_sequence_number = self.get_success(
-            self.state_deletion_store.get_pending_deletions([state_group])
-        )
-
-        can_be_deleted = self.get_success(
-            self.state_deletion_store.db_pool.runInteraction(
-                "test_existing_pending_deletion_is_cleared",
-                self.state_deletion_store.get_state_groups_ready_for_potential_deletion_txn,
-                state_group_to_sequence_number,
-            )
-        )
-
-        return state_group in can_be_deleted
-
-    def test_no_deletion(self) -> None:
-        """Test that calling persisting_state_group_references is fine if
-        nothing is pending deletion"""
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                sender=self.user_id,
-            )
-        )
-
-        ctx_mgr = self.state_deletion_store.persisting_state_group_references(
-            [(event, context)]
-        )
-
-        self.get_success(ctx_mgr.__aenter__())
-        self.get_success(ctx_mgr.__aexit__(None, None, None))
-
-    def test_no_deletion_error(self) -> None:
-        """Test that calling persisting_state_group_references is fine if
-        nothing is pending deletion, but an error occurs."""
-
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                sender=self.user_id,
-            )
-        )
-
-        ctx_mgr = self.state_deletion_store.persisting_state_group_references(
-            [(event, context)]
-        )
-
-        self.get_success(ctx_mgr.__aenter__())
-        self.get_success(ctx_mgr.__aexit__(Exception, Exception("test"), None))
-
-    def test_existing_pending_deletion_is_cleared(self) -> None:
-        """Test that the pending deletion flag gets cleared when the state group
-        gets persisted."""
-
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                state_key="",
-                sender=self.user_id,
-            )
-        )
-        assert context.state_group is not None
-
-        # Mark a state group that we're referencing as pending deletion.
-        self.get_success(
-            self.state_deletion_store.mark_state_groups_as_pending_deletion(
-                [context.state_group]
-            )
-        )
-
-        ctx_mgr = self.state_deletion_store.persisting_state_group_references(
-            [(event, context)]
-        )
-
-        self.get_success(ctx_mgr.__aenter__())
-        self.get_success(ctx_mgr.__aexit__(None, None, None))
-
-        # The pending deletion flag should be cleared
-        pending_deletion = self.get_success(
-            self.state_deletion_store.db_pool.simple_select_one_onecol(
-                table="state_groups_pending_deletion",
-                keyvalues={"state_group": context.state_group},
-                retcol="1",
-                allow_none=True,
-                desc="test_existing_pending_deletion_is_cleared",
-            )
-        )
-        self.assertIsNone(pending_deletion)
-
-    def test_pending_deletion_is_cleared_during_persist(self) -> None:
-        """Test that the pending deletion flag is cleared when a state group
-        gets marked for deletion during persistence"""
-
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                state_key="",
-                sender=self.user_id,
-            )
-        )
-        assert context.state_group is not None
-
-        ctx_mgr = self.state_deletion_store.persisting_state_group_references(
-            [(event, context)]
-        )
-        self.get_success(ctx_mgr.__aenter__())
-
-        # Mark the state group that we're referencing as pending deletion,
-        # *after* we have started persisting.
-        self.get_success(
-            self.state_deletion_store.mark_state_groups_as_pending_deletion(
-                [context.state_group]
-            )
-        )
-
-        self.get_success(ctx_mgr.__aexit__(None, None, None))
-
-        # The pending deletion flag should be cleared
-        pending_deletion = self.get_success(
-            self.state_deletion_store.db_pool.simple_select_one_onecol(
-                table="state_groups_pending_deletion",
-                keyvalues={"state_group": context.state_group},
-                retcol="1",
-                allow_none=True,
-                desc="test_existing_pending_deletion_is_cleared",
-            )
-        )
-        self.assertIsNone(pending_deletion)
-
-    def test_deletion_check(self) -> None:
-        """Test that the `get_state_groups_that_can_be_purged_txn` check is
-        correct during different points of the lifecycle of persisting an
-        event."""
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                state_key="",
-                sender=self.user_id,
-            )
-        )
-        assert context.state_group is not None
-
-        self.get_success(
-            self.state_deletion_store.mark_state_groups_as_pending_deletion(
-                [context.state_group]
-            )
-        )
-
-        # We shouldn't be able to delete the state group as not enough time as passed
-        can_be_deleted = self.check_if_can_be_deleted(context.state_group)
-        self.assertFalse(can_be_deleted)
-
-        # After enough time we can delete the state group
-        self.reactor.advance(
-            1 + self.state_deletion_store.DELAY_BEFORE_DELETION_MS / 1000
-        )
-        can_be_deleted = self.check_if_can_be_deleted(context.state_group)
-        self.assertTrue(can_be_deleted)
-
-        ctx_mgr = self.state_deletion_store.persisting_state_group_references(
-            [(event, context)]
-        )
-        self.get_success(ctx_mgr.__aenter__())
-
-        # But once we start persisting we can't delete the state group
-        can_be_deleted = self.check_if_can_be_deleted(context.state_group)
-        self.assertFalse(can_be_deleted)
-
-        self.get_success(ctx_mgr.__aexit__(None, None, None))
-
-        # The pending deletion flag should remain cleared after persistence has
-        # finished.
-        can_be_deleted = self.check_if_can_be_deleted(context.state_group)
-        self.assertFalse(can_be_deleted)
-
-    def test_deletion_error_during_persistence(self) -> None:
-        """Test that state groups remain marked as pending deletion if persisting
-        the event fails."""
-
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                state_key="",
-                sender=self.user_id,
-            )
-        )
-        assert context.state_group is not None
-
-        # Mark a state group that we're referencing as pending deletion.
-        self.get_success(
-            self.state_deletion_store.mark_state_groups_as_pending_deletion(
-                [context.state_group]
-            )
-        )
-
-        ctx_mgr = self.state_deletion_store.persisting_state_group_references(
-            [(event, context)]
-        )
-
-        self.get_success(ctx_mgr.__aenter__())
-        self.get_success(ctx_mgr.__aexit__(Exception, Exception("test"), None))
-
-        # We should be able to delete the state group after a certain amount of
-        # time
-        self.reactor.advance(
-            1 + self.state_deletion_store.DELAY_BEFORE_DELETION_MS / 1000
-        )
-        can_be_deleted = self.check_if_can_be_deleted(context.state_group)
-        self.assertTrue(can_be_deleted)
-
-    def test_race_between_check_and_insert(self) -> None:
-        """Check that we correctly handle the race where we go to delete a
-        state group, check that it is unreferenced, and then it becomes
-        referenced just before we delete it."""
-
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                state_key="",
-                sender=self.user_id,
-            )
-        )
-        assert context.state_group is not None
-
-        # Mark a state group that we're referencing as pending deletion.
-        self.get_success(
-            self.state_deletion_store.mark_state_groups_as_pending_deletion(
-                [context.state_group]
-            )
-        )
-
-        # Advance time enough so we can delete the state group
-        self.reactor.advance(
-            1 + self.state_deletion_store.DELAY_BEFORE_DELETION_MS / 1000
-        )
-
-        # Check that we'd be able to delete this state group.
-        state_group_to_sequence_number = self.get_success(
-            self.state_deletion_store.get_pending_deletions([context.state_group])
-        )
-
-        can_be_deleted = self.get_success(
-            self.state_deletion_store.db_pool.runInteraction(
-                "test_existing_pending_deletion_is_cleared",
-                self.state_deletion_store.get_state_groups_ready_for_potential_deletion_txn,
-                state_group_to_sequence_number,
-            )
-        )
-        self.assertCountEqual(can_be_deleted, [context.state_group])
-
-        # ... in the real world we'd check that the state group isn't referenced here ...
-
-        # Now we persist the event to reference the state group, *after* we
-        # check that the state group wasn't referenced
-        ctx_mgr = self.state_deletion_store.persisting_state_group_references(
-            [(event, context)]
-        )
-
-        self.get_success(ctx_mgr.__aenter__())
-        self.get_success(ctx_mgr.__aexit__(Exception, Exception("test"), None))
-
-        # We simulate a pause (required to hit the race)
-        self.reactor.advance(
-            1 + self.state_deletion_store.DELAY_BEFORE_DELETION_MS / 1000
-        )
-
-        # We should no longer be able to delete the state group, without having
-        # to recheck if its referenced.
-        can_be_deleted = self.get_success(
-            self.state_deletion_store.db_pool.runInteraction(
-                "test_existing_pending_deletion_is_cleared",
-                self.state_deletion_store.get_state_groups_ready_for_potential_deletion_txn,
-                state_group_to_sequence_number,
-            )
-        )
-        self.assertCountEqual(can_be_deleted, [])
-
-    def test_remove_ancestors_from_can_delete(self) -> None:
-        """Test that if a state group is not ready to be deleted, we also don't
-        delete anything that is referenced by it"""
-
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                state_key="",
-                sender=self.user_id,
-            )
-        )
-        assert context.state_group is not None
-
-        # Create a new state group that references the one from the event
-        new_state_group = self.get_success(
-            self.state_store.store_state_group(
-                event.event_id,
-                event.room_id,
-                prev_group=context.state_group,
-                delta_ids={},
-                current_state_ids=None,
-            )
-        )
-
-        # Mark them both as pending deletion
-        self.get_success(
-            self.state_deletion_store.mark_state_groups_as_pending_deletion(
-                [context.state_group, new_state_group]
-            )
-        )
-
-        # Advance time enough so we can delete the state group so they're both
-        # ready for deletion.
-        self.reactor.advance(
-            1 + self.state_deletion_store.DELAY_BEFORE_DELETION_MS / 1000
-        )
-
-        # We can now delete both state groups
-        self.assertTrue(self.check_if_can_be_deleted(context.state_group))
-        self.assertTrue(self.check_if_can_be_deleted(new_state_group))
-
-        # Use the new_state_group to bump its deletion time
-        self.get_success(
-            self.state_store.store_state_group(
-                event.event_id,
-                event.room_id,
-                prev_group=new_state_group,
-                delta_ids={},
-                current_state_ids=None,
-            )
-        )
-
-        # We should now not be able to delete either of the state groups.
-        state_group_to_sequence_number = self.get_success(
-            self.state_deletion_store.get_pending_deletions(
-                [context.state_group, new_state_group]
-            )
-        )
-
-        # We shouldn't be able to delete the state group as not enough time has passed
-        can_be_deleted = self.get_success(
-            self.state_deletion_store.db_pool.runInteraction(
-                "test_existing_pending_deletion_is_cleared",
-                self.state_deletion_store.get_state_groups_ready_for_potential_deletion_txn,
-                state_group_to_sequence_number,
-            )
-        )
-        self.assertCountEqual(can_be_deleted, [])
-
-    def test_newly_referenced_state_group_gets_removed_from_pending(self) -> None:
-        """Check that if a state group marked for deletion becomes referenced
-        (without being removed from pending deletion table), it gets removed
-        from pending deletion table."""
-
-        event, context = self.get_success(
-            create_event(
-                self.hs,
-                room_id=self.room_id,
-                type="m.test",
-                state_key="",
-                sender=self.user_id,
-            )
-        )
-        assert context.state_group is not None
-
-        # Mark a state group that we're referencing as pending deletion.
-        self.get_success(
-            self.state_deletion_store.mark_state_groups_as_pending_deletion(
-                [context.state_group]
-            )
-        )
-
-        # Advance time enough so we can delete the state group so they're both
-        # ready for deletion.
-        self.reactor.advance(
-            1 + self.state_deletion_store.DELAY_BEFORE_DELETION_MS / 1000
-        )
-
-        # Manually insert into the table to mimic the state group getting used.
-        self.get_success(
-            self.store.db_pool.simple_insert(
-                table="event_to_state_groups",
-                values={"state_group": context.state_group, "event_id": event.event_id},
-                desc="test_newly_referenced_state_group_gets_removed_from_pending",
-            )
-        )
-
-        # Manually run the background task to delete pending state groups.
-        self.get_success(self.purge_events._delete_state_groups_loop())
-
-        # The pending deletion flag should be cleared...
-        pending_deletion = self.get_success(
-            self.state_deletion_store.db_pool.simple_select_one_onecol(
-                table="state_groups_pending_deletion",
-                keyvalues={"state_group": context.state_group},
-                retcol="1",
-                allow_none=True,
-                desc="test_newly_referenced_state_group_gets_removed_from_pending",
-            )
-        )
-        self.assertIsNone(pending_deletion)
-
-        # .. but the state should not have been deleted.
-        state = self.get_success(
-            self.state_store._get_state_for_groups([context.state_group])
-        )
-        self.assertGreater(len(state[context.state_group]), 0)

tests/test_state.py

@@ -31,7 +31,7 @@ from typing import (
     Tuple,
     cast,
 )
-from unittest.mock import AsyncMock, Mock
+from unittest.mock import Mock
 
 from twisted.internet import defer
 
@@ -221,16 +221,7 @@ class Graph:
 class StateTestCase(unittest.TestCase):
     def setUp(self) -> None:
         self.dummy_store = _DummyStore()
-
-        # Add a dummy epoch store that always retruns that we have all the
-        # necessary state groups.
-        dummy_deletion_store = AsyncMock()
-        dummy_deletion_store.check_state_groups_and_bump_deletion.return_value = []
-
-        storage_controllers = Mock(
-            main=self.dummy_store,
-            state=self.dummy_store,
-        )
+        storage_controllers = Mock(main=self.dummy_store, state=self.dummy_store)
         hs = Mock(
             spec_set=[
                 "config",
@@ -250,10 +241,7 @@ class StateTestCase(unittest.TestCase):
         )
         clock = cast(Clock, MockClock())
         hs.config = default_config("tesths", True)
-        hs.get_datastores.return_value = Mock(
-            main=self.dummy_store,
-            state_deletion=dummy_deletion_store,
-        )
+        hs.get_datastores.return_value = Mock(main=self.dummy_store)
         hs.get_state_handler.return_value = None
         hs.get_clock.return_value = clock
         hs.get_macaroon_generator.return_value = MacaroonGenerator(