diff --git a/.buildkite/docker-compose.py27.pg94.yaml b/.buildkite/docker-compose.py27.pg94.yaml
deleted file mode 100644
index 2d4b9eadd9..0000000000
--- a/.buildkite/docker-compose.py27.pg94.yaml
+++ /dev/null
@@ -1,21 +0,0 @@
-version: '3.1'
-
-services:
-
- postgres:
- image: postgres:9.4
- environment:
- POSTGRES_PASSWORD: postgres
-
- testenv:
- image: python:2.7
- depends_on:
- - postgres
- env_file: .env
- environment:
- SYNAPSE_POSTGRES_HOST: postgres
- SYNAPSE_POSTGRES_USER: postgres
- SYNAPSE_POSTGRES_PASSWORD: postgres
- working_dir: /app
- volumes:
- - ..:/app
diff --git a/.buildkite/docker-compose.py27.pg95.yaml b/.buildkite/docker-compose.py27.pg95.yaml
deleted file mode 100644
index c6a41f1da0..0000000000
--- a/.buildkite/docker-compose.py27.pg95.yaml
+++ /dev/null
@@ -1,21 +0,0 @@
-version: '3.1'
-
-services:
-
- postgres:
- image: postgres:9.5
- environment:
- POSTGRES_PASSWORD: postgres
-
- testenv:
- image: python:2.7
- depends_on:
- - postgres
- env_file: .env
- environment:
- SYNAPSE_POSTGRES_HOST: postgres
- SYNAPSE_POSTGRES_USER: postgres
- SYNAPSE_POSTGRES_PASSWORD: postgres
- working_dir: /app
- volumes:
- - ..:/app
diff --git a/.buildkite/pipeline.yml b/.buildkite/pipeline.yml
index 44b258dca6..9135c95a87 100644
--- a/.buildkite/pipeline.yml
+++ b/.buildkite/pipeline.yml
@@ -48,17 +48,6 @@ steps:
- command:
- "python -m pip install tox"
- - "tox -e py27,codecov"
- label: ":python: 2.7 / SQLite"
- env:
- TRIAL_FLAGS: "-j 2"
- plugins:
- - docker#v3.0.1:
- image: "python:2.7"
- propagate-environment: true
-
- - command:
- - "python -m pip install tox"
- "tox -e py35,codecov"
label: ":python: 3.5 / SQLite"
env:
@@ -90,39 +79,6 @@ steps:
image: "python:3.7"
propagate-environment: true
- - command:
- - "python -m pip install tox"
- - "tox -e py27-old,codecov"
- label: ":python: 2.7 / SQLite / Old Deps"
- env:
- TRIAL_FLAGS: "-j 2"
- plugins:
- - docker#v3.0.1:
- image: "python:2.7"
- propagate-environment: true
-
- - label: ":python: 2.7 / :postgres: 9.4"
- env:
- TRIAL_FLAGS: "-j 4"
- command:
- - "bash -c 'python -m pip install tox && python -m tox -e py27-postgres,codecov'"
- plugins:
- - docker-compose#v2.1.0:
- run: testenv
- config:
- - .buildkite/docker-compose.py27.pg94.yaml
-
- - label: ":python: 2.7 / :postgres: 9.5"
- env:
- TRIAL_FLAGS: "-j 4"
- command:
- - "bash -c 'python -m pip install tox && python -m tox -e py27-postgres,codecov'"
- plugins:
- - docker-compose#v2.1.0:
- run: testenv
- config:
- - .buildkite/docker-compose.py27.pg95.yaml
-
- label: ":python: 3.5 / :postgres: 9.4"
env:
TRIAL_FLAGS: "-j 4"
diff --git a/.circleci/config.yml b/.circleci/config.yml
index 137747dae3..6775a554a7 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -10,6 +10,13 @@ jobs:
- run: docker push matrixdotorg/synapse:${CIRCLE_TAG}
- run: docker push matrixdotorg/synapse:${CIRCLE_TAG}-py2
- run: docker push matrixdotorg/synapse:${CIRCLE_TAG}-py3
+ dockerhubuploadreleasebranch:
+ machine: true
+ steps:
+ - checkout
+ - run: docker build -f docker/Dockerfile --label gitsha1=${CIRCLE_SHA1} -t matrixdotorg/synapse:${CIRCLE_BRANCH} --build-arg PYTHON_VERSION=3.7 .
+ - run: docker login --username $DOCKER_HUB_USERNAME --password $DOCKER_HUB_PASSWORD
+ - run: docker push matrixdotorg/synapse:${CIRCLE_BRANCH}
dockerhubuploadlatest:
machine: true
steps:
@@ -20,56 +27,6 @@ jobs:
- run: docker push matrixdotorg/synapse:latest
- run: docker push matrixdotorg/synapse:latest-py2
- run: docker push matrixdotorg/synapse:latest-py3
- sytestpy2:
- docker:
- - image: matrixdotorg/sytest-synapsepy2
- working_directory: /src
- steps:
- - checkout
- - run: /synapse_sytest.sh
- - store_artifacts:
- path: /logs
- destination: logs
- - store_test_results:
- path: /logs
- sytestpy2postgres:
- docker:
- - image: matrixdotorg/sytest-synapsepy2
- working_directory: /src
- steps:
- - checkout
- - run: POSTGRES=1 /synapse_sytest.sh
- - store_artifacts:
- path: /logs
- destination: logs
- - store_test_results:
- path: /logs
- sytestpy2merged:
- docker:
- - image: matrixdotorg/sytest-synapsepy2
- working_directory: /src
- steps:
- - checkout
- - run: bash .circleci/merge_base_branch.sh
- - run: /synapse_sytest.sh
- - store_artifacts:
- path: /logs
- destination: logs
- - store_test_results:
- path: /logs
- sytestpy2postgresmerged:
- docker:
- - image: matrixdotorg/sytest-synapsepy2
- working_directory: /src
- steps:
- - checkout
- - run: bash .circleci/merge_base_branch.sh
- - run: POSTGRES=1 /synapse_sytest.sh
- - store_artifacts:
- path: /logs
- destination: logs
- - store_test_results:
- path: /logs
sytestpy3:
docker:
@@ -126,30 +83,6 @@ workflows:
version: 2
build:
jobs:
- - sytestpy2:
- filters:
- branches:
- only: /develop|master|release-.*/
- - sytestpy2postgres:
- filters:
- branches:
- only: /develop|master|release-.*/
- - sytestpy3:
- filters:
- branches:
- only: /develop|master|release-.*/
- - sytestpy3postgres:
- filters:
- branches:
- only: /develop|master|release-.*/
- - sytestpy2merged:
- filters:
- branches:
- ignore: /develop|master|release-.*/
- - sytestpy2postgresmerged:
- filters:
- branches:
- ignore: /develop|master|release-.*/
- sytestpy3merged:
filters:
branches:
@@ -164,6 +97,10 @@ workflows:
only: /v[0-9].[0-9]+.[0-9]+.*/
branches:
ignore: /.*/
+ - dockerhubuploadreleasebranch:
+ filters:
+ branches:
+ only: shhs
- dockerhubuploadlatest:
filters:
branches:
diff --git a/.gitignore b/.gitignore
index a84c41b0c9..702dd33eb1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -19,6 +19,7 @@ _trial_temp*/
/*.signing.key
/env/
/homeserver*.yaml
+/logs
/media_store/
/uploads
@@ -37,4 +38,3 @@ _trial_temp*/
/docs/build/
/htmlcov
/pip-wheel-metadata/
-
diff --git a/changelog.d/4338.feature b/changelog.d/4338.feature
new file mode 100644
index 0000000000..01285e965c
--- /dev/null
+++ b/changelog.d/4338.feature
@@ -0,0 +1 @@
+Synapse now more efficiently collates room statistics.
diff --git a/changelog.d/5072.feature b/changelog.d/5072.feature
new file mode 100644
index 0000000000..99fda5616a
--- /dev/null
+++ b/changelog.d/5072.feature
@@ -0,0 +1 @@
+Synapse can now be configured to not join remote rooms of a given "complexity" (currently, state events). This option can be used to prevent adverse performance on resource-constrained homeservers.
\ No newline at end of file
diff --git a/changelog.d/5099.misc b/changelog.d/5099.misc
new file mode 100644
index 0000000000..21863e4e3e
--- /dev/null
+++ b/changelog.d/5099.misc
@@ -0,0 +1 @@
+Python 2 has been removed from the CI.
diff --git a/docker/start.py b/docker/start.py
index 2da555272a..6affd06d94 100755
--- a/docker/start.py
+++ b/docker/start.py
@@ -47,6 +47,11 @@ if mode == "generate":
# In normal mode, generate missing keys if any, then run synapse
else:
+
+ # tell synapse to put any generated keys in /data rather than /compiled by
+ # default
+ SYNAPSE_KEY_PATH = environ.get("SYNAPSE_KEY_PATH", "/data")
+
if "SYNAPSE_CONFIG_PATH" in environ:
config_path = environ["SYNAPSE_CONFIG_PATH"]
else:
@@ -59,7 +64,7 @@ else:
if not os.path.exists("/compiled"): os.mkdir("/compiled")
config_path = "/compiled/homeserver.yaml"
-
+
# Convert SYNAPSE_NO_TLS to boolean if exists
if "SYNAPSE_NO_TLS" in environ:
tlsanswerstring = str.lower(environ["SYNAPSE_NO_TLS"])
@@ -79,9 +84,7 @@ else:
args += [
"--config-path", config_path,
-
- # tell synapse to put any generated keys in /data rather than /compiled
- "--keys-directory", "/data",
+ "--keys-directory", SYNAPSE_KEY_PATH
]
# Generate missing keys and start synapse
diff --git a/docs/sample_config.yaml b/docs/sample_config.yaml
index f658ec8ecd..6c80d376f9 100644
--- a/docs/sample_config.yaml
+++ b/docs/sample_config.yaml
@@ -271,6 +271,17 @@ listeners:
# Used by phonehome stats to group together related servers.
#server_context: context
+# Resource-constrained Homeserver Settings
+#
+# If limit_large_remote_room_joins is True, the room complexity will be
+# checked before a user joins a new remote room. If it is above
+# limit_large_remote_room_complexity, it will disallow joining or
+# instantly leave.
+#
+# Uncomment the below lines to enable:
+#limit_large_remote_room_joins: True
+#limit_large_remote_room_complexity: 1.0
+
# Whether to require a user to be in the room to add an alias to it.
# Defaults to 'true'.
#
@@ -1153,6 +1164,22 @@ password_config:
#
+
+# Local statistics collection. Used in populating the room directory.
+#
+# 'bucket_size' controls how large each statistics timeslice is. It can
+# be defined in a human readable short form -- e.g. "1d", "1y".
+#
+# 'retention' controls how long historical statistics will be kept for.
+# It can be defined in a human readable short form -- e.g. "1d", "1y".
+#
+#
+#stats:
+# enabled: true
+# bucket_size: 1d
+# retention: 1y
+
+
# Server Notices room configuration
#
# Uncomment this section to enable a room which can be used to send notices
diff --git a/synapse/api/constants.py b/synapse/api/constants.py
index 6b347b1749..ee129c8689 100644
--- a/synapse/api/constants.py
+++ b/synapse/api/constants.py
@@ -79,6 +79,7 @@ class EventTypes(object):
RoomHistoryVisibility = "m.room.history_visibility"
CanonicalAlias = "m.room.canonical_alias"
+ Encryption = "m.room.encryption"
RoomAvatar = "m.room.avatar"
RoomEncryption = "m.room.encryption"
GuestAccess = "m.room.guest_access"
diff --git a/synapse/api/urls.py b/synapse/api/urls.py
index 3c6bddff7a..e16c386a14 100644
--- a/synapse/api/urls.py
+++ b/synapse/api/urls.py
@@ -26,6 +26,7 @@ CLIENT_API_PREFIX = "/_matrix/client"
FEDERATION_PREFIX = "/_matrix/federation"
FEDERATION_V1_PREFIX = FEDERATION_PREFIX + "/v1"
FEDERATION_V2_PREFIX = FEDERATION_PREFIX + "/v2"
+FEDERATION_UNSTABLE_PREFIX = FEDERATION_PREFIX + "/unstable"
STATIC_PREFIX = "/_matrix/static"
WEB_CLIENT_PREFIX = "/_matrix/client"
CONTENT_REPO_PREFIX = "/_matrix/content"
diff --git a/synapse/config/homeserver.py b/synapse/config/homeserver.py
index 727fdc54d8..5c4fc8ff21 100644
--- a/synapse/config/homeserver.py
+++ b/synapse/config/homeserver.py
@@ -13,6 +13,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
+
from .api import ApiConfig
from .appservice import AppServiceConfig
from .captcha import CaptchaConfig
@@ -36,20 +37,41 @@ from .saml2_config import SAML2Config
from .server import ServerConfig
from .server_notices_config import ServerNoticesConfig
from .spam_checker import SpamCheckerConfig
+from .stats import StatsConfig
from .tls import TlsConfig
from .user_directory import UserDirectoryConfig
from .voip import VoipConfig
from .workers import WorkerConfig
-class HomeServerConfig(ServerConfig, TlsConfig, DatabaseConfig, LoggingConfig,
- RatelimitConfig, ContentRepositoryConfig, CaptchaConfig,
- VoipConfig, RegistrationConfig, MetricsConfig, ApiConfig,
- AppServiceConfig, KeyConfig, SAML2Config, CasConfig,
- JWTConfig, PasswordConfig, EmailConfig,
- WorkerConfig, PasswordAuthProviderConfig, PushConfig,
- SpamCheckerConfig, GroupsConfig, UserDirectoryConfig,
- ConsentConfig,
- ServerNoticesConfig, RoomDirectoryConfig,
- ):
+class HomeServerConfig(
+ ServerConfig,
+ TlsConfig,
+ DatabaseConfig,
+ LoggingConfig,
+ RatelimitConfig,
+ ContentRepositoryConfig,
+ CaptchaConfig,
+ VoipConfig,
+ RegistrationConfig,
+ MetricsConfig,
+ ApiConfig,
+ AppServiceConfig,
+ KeyConfig,
+ SAML2Config,
+ CasConfig,
+ JWTConfig,
+ PasswordConfig,
+ EmailConfig,
+ WorkerConfig,
+ PasswordAuthProviderConfig,
+ PushConfig,
+ SpamCheckerConfig,
+ GroupsConfig,
+ UserDirectoryConfig,
+ ConsentConfig,
+ StatsConfig,
+ ServerNoticesConfig,
+ RoomDirectoryConfig,
+):
pass
diff --git a/synapse/config/repository.py b/synapse/config/repository.py
index fbfcecc240..db7055ac5e 100644
--- a/synapse/config/repository.py
+++ b/synapse/config/repository.py
@@ -107,11 +107,18 @@ def parse_thumbnail_requirements(thumbnail_sizes):
class ContentRepositoryConfig(Config):
def read_config(self, config):
+ self.enable_media_repo = config.get("enable_media_repo", True)
+
self.max_upload_size = self.parse_size(config.get("max_upload_size", "10M"))
self.max_image_pixels = self.parse_size(config.get("max_image_pixels", "32M"))
self.max_spider_size = self.parse_size(config.get("max_spider_size", "10M"))
- self.media_store_path = self.ensure_directory(config["media_store_path"])
+ if self.enable_media_repo:
+ self.media_store_path = self.ensure_directory(config["media_store_path"])
+ self.uploads_path = self.ensure_directory(config["uploads_path"])
+ else:
+ self.media_store_path = None
+ self.uploads_path = None
backup_media_store_path = config.get("backup_media_store_path")
@@ -168,7 +175,6 @@ class ContentRepositoryConfig(Config):
(provider_class, parsed_config, wrapper_config,)
)
- self.uploads_path = self.ensure_directory(config["uploads_path"])
self.dynamic_thumbnails = config.get("dynamic_thumbnails", False)
self.thumbnail_requirements = parse_thumbnail_requirements(
config.get("thumbnail_sizes", DEFAULT_THUMBNAIL_SIZES),
diff --git a/synapse/config/server.py b/synapse/config/server.py
index f34aa42afa..30adb5c5e7 100644
--- a/synapse/config/server.py
+++ b/synapse/config/server.py
@@ -221,6 +221,12 @@ class ServerConfig(Config):
self.gc_thresholds = read_gc_thresholds(config.get("gc_thresholds", None))
+ # Resource-constrained Homeserver Configuration
+ self.limit_large_room_joins = config.get("limit_large_remote_room_joins", False)
+ self.limit_large_room_complexity = config.get(
+ "limit_large_remote_room_complexity", 1.0
+ )
+
bind_port = config.get("bind_port")
if bind_port:
if config.get("no_tls", False):
@@ -572,6 +578,17 @@ class ServerConfig(Config):
# Used by phonehome stats to group together related servers.
#server_context: context
+ # Resource-constrained Homeserver Settings
+ #
+ # If limit_large_remote_room_joins is True, the room complexity will be
+ # checked before a user joins a new remote room. If it is above
+ # limit_large_remote_room_complexity, it will disallow joining or
+ # instantly leave.
+ #
+ # Uncomment the below lines to enable:
+ #limit_large_remote_room_joins: True
+ #limit_large_remote_room_complexity: 1.0
+
# Whether to require a user to be in the room to add an alias to it.
# Defaults to 'true'.
#
diff --git a/synapse/config/stats.py b/synapse/config/stats.py
new file mode 100644
index 0000000000..80fc1b9dd0
--- /dev/null
+++ b/synapse/config/stats.py
@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018 New Vector Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import division
+
+import sys
+
+from ._base import Config
+
+
+class StatsConfig(Config):
+ """Stats Configuration
+ Configuration for the behaviour of synapse's stats engine
+ """
+
+ def read_config(self, config):
+ self.stats_enabled = True
+ self.stats_bucket_size = 86400
+ self.stats_retention = sys.maxsize
+ stats_config = config.get("stats", None)
+ if stats_config:
+ self.stats_enabled = stats_config.get("enabled", self.stats_enabled)
+ self.stats_bucket_size = (
+ self.parse_duration(stats_config.get("bucket_size", "1d")) / 1000
+ )
+ self.stats_retention = (
+ self.parse_duration(
+ stats_config.get("retention", "%ds" % (sys.maxsize,))
+ )
+ / 1000
+ )
+
+ def default_config(self, config_dir_path, server_name, **kwargs):
+ return """
+ # Local statistics collection. Used in populating the room directory.
+ #
+ # 'bucket_size' controls how large each statistics timeslice is. It can
+ # be defined in a human readable short form -- e.g. "1d", "1y".
+ #
+ # 'retention' controls how long historical statistics will be kept for.
+ # It can be defined in a human readable short form -- e.g. "1d", "1y".
+ #
+ #
+ #stats:
+ # enabled: true
+ # bucket_size: 1d
+ # retention: 1y
+ """
diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index f3fc897a0a..d4c62ec0c8 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -992,3 +992,39 @@ class FederationClient(FederationBase):
)
raise RuntimeError("Failed to send to any server.")
+
+ @defer.inlineCallbacks
+ def get_room_complexity(self, destination, room_id):
+ """
+ Fetch the complexity of a remote room from another server.
+
+ Args:
+ destination (str): The remote server
+ room_id (str): The room ID to ask about.
+
+ Returns:
+ Deferred[dict] or Deferred[None]: Dict contains the complexity
+ metric versions, while None means we could not fetch the complexity.
+ """
+ try:
+ complexity = yield self.transport_layer.get_room_complexity(
+ destination=destination,
+ room_id=room_id
+ )
+ defer.returnValue(complexity)
+ except CodeMessageException as e:
+ # We didn't manage to get it -- probably a 404. We are okay if other
+ # servers don't give it to us.
+ logger.debug(
+ "Failed to fetch room complexity via %s for %s, got a %d",
+ destination, room_id, e.code
+ )
+ except Exception:
+ logger.exception(
+ "Failed to fetch room complexity via %s for %s",
+ destination, room_id
+ )
+
+ # If we don't manage to find it, return None. It's not an error if a
+ # server doesn't give it to us.
+ defer.returnValue(None)
diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py
index e424c40fdf..2797122700 100644
--- a/synapse/federation/transport/client.py
+++ b/synapse/federation/transport/client.py
@@ -21,7 +21,11 @@ from six.moves import urllib
from twisted.internet import defer
from synapse.api.constants import Membership
-from synapse.api.urls import FEDERATION_V1_PREFIX, FEDERATION_V2_PREFIX
+from synapse.api.urls import (
+ FEDERATION_UNSTABLE_PREFIX,
+ FEDERATION_V1_PREFIX,
+ FEDERATION_V2_PREFIX,
+)
from synapse.util.logutils import log_function
logger = logging.getLogger(__name__)
@@ -959,6 +963,28 @@ class TransportLayerClient(object):
ignore_backoff=True,
)
+ def get_room_complexity(self, destination, room_id):
+ """
+ Args:
+ destination (str): The remote server
+ room_id (str): The room ID to ask about.
+ """
+ path = _create_path(
+ FEDERATION_UNSTABLE_PREFIX, "/rooms/%s/complexity", room_id
+ )
+
+ return self.client.get_json(
+ destination=destination,
+ path=path
+ )
+
+
+def _create_path(federation_prefix, path, *args):
+ """
+ Ensures that all args are url encoded.
+ """
+ return federation_prefix + path % tuple(urllib.parse.quote(arg, "") for arg in args)
+
def _create_v1_path(path, *args):
"""Creates a path against V1 federation API from the path template and
@@ -975,10 +1001,7 @@ def _create_v1_path(path, *args):
Returns:
str
"""
- return (
- FEDERATION_V1_PREFIX
- + path % tuple(urllib.parse.quote(arg, "") for arg in args)
- )
+ return _create_path(FEDERATION_V1_PREFIX, path, *args)
def _create_v2_path(path, *args):
@@ -996,7 +1019,4 @@ def _create_v2_path(path, *args):
Returns:
str
"""
- return (
- FEDERATION_V2_PREFIX
- + path % tuple(urllib.parse.quote(arg, "") for arg in args)
- )
+ return _create_path(FEDERATION_V2_PREFIX, path, *args)
diff --git a/synapse/federation/transport/server.py b/synapse/federation/transport/server.py
index 385eda2dca..db73385ccd 100644
--- a/synapse/federation/transport/server.py
+++ b/synapse/federation/transport/server.py
@@ -23,7 +23,11 @@ from twisted.internet import defer
import synapse
from synapse.api.errors import Codes, FederationDeniedError, SynapseError
from synapse.api.room_versions import RoomVersions
-from synapse.api.urls import FEDERATION_V1_PREFIX, FEDERATION_V2_PREFIX
+from synapse.api.urls import (
+ FEDERATION_UNSTABLE_PREFIX,
+ FEDERATION_V1_PREFIX,
+ FEDERATION_V2_PREFIX,
+)
from synapse.http.endpoint import parse_and_validate_server_name
from synapse.http.server import JsonResource
from synapse.http.servlet import (
@@ -1304,6 +1308,26 @@ class FederationGroupsSettingJoinPolicyServlet(BaseFederationServlet):
defer.returnValue((200, new_content))
+class RoomComplexityServlet(BaseFederationServlet):
+ PATH = "/rooms/(?P<room_id>[^/]*)/complexity"
+ PREFIX = FEDERATION_UNSTABLE_PREFIX
+
+ @defer.inlineCallbacks
+ def on_GET(self, origin, content, query, room_id):
+
+ store = self.handler.hs.get_datastore()
+
+ is_public = yield store.is_room_world_readable_or_publicly_joinable(
+ room_id
+ )
+
+ if not is_public:
+ raise SynapseError(404, "Room not found", errcode=Codes.INVALID_PARAM)
+
+ complexity = yield store.get_room_complexity(room_id)
+ defer.returnValue((200, complexity))
+
+
FEDERATION_SERVLET_CLASSES = (
FederationSendServlet,
FederationEventServlet,
@@ -1327,6 +1351,7 @@ FEDERATION_SERVLET_CLASSES = (
FederationThirdPartyInviteExchangeServlet,
On3pidBindServlet,
FederationVersionServlet,
+ RoomComplexityServlet,
)
OPENID_SERVLET_CLASSES = (
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 2202ed699a..e2f5b23e33 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -2729,3 +2729,28 @@ class FederationHandler(BaseHandler):
)
else:
return user_joined_room(self.distributor, user, room_id)
+
+ @defer.inlineCallbacks
+ def get_room_complexity(self, remote_room_hosts, room_id):
+ """
+ Fetch the complexity of a remote room over federation.
+
+ Args:
+ remote_room_hosts (list[str]): The remote servers to ask.
+ room_id (str): The room ID to ask about.
+
+ Returns:
+ Deferred[dict] or Deferred[None]: Dict contains the complexity
+ metric versions, while None means we could not fetch the complexity.
+ """
+
+ for host in remote_room_hosts:
+ res = yield self.federation_client.get_room_complexity(host, room_id)
+
+ # We got a result, return it.
+ if res:
+ defer.returnValue(res)
+
+ # We fell off the bottom, couldn't get the complexity from anyone. Oh
+ # well.
+ defer.returnValue(None)
diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index 93ac986c86..7aefbc426a 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -26,8 +26,7 @@ from unpaddedbase64 import decode_base64
from twisted.internet import defer
-import synapse.server
-import synapse.types
+from synapse import types
from synapse.api.constants import EventTypes, Membership
from synapse.api.errors import AuthError, Codes, SynapseError
from synapse.types import RoomID, UserID
@@ -590,7 +589,7 @@ class RoomMemberHandler(object):
)
assert self.hs.is_mine(sender), "Sender must be our own: %s" % (sender,)
else:
- requester = synapse.types.create_requester(target_user)
+ requester = types.create_requester(target_user)
prev_event = yield self.event_creation_handler.deduplicate_state_event(
event, context,
@@ -1018,13 +1017,53 @@ class RoomMemberMasterHandler(RoomMemberHandler):
self.distributor.declare("user_left_room")
@defer.inlineCallbacks
+ def _is_remote_room_too_complex(self, room_id, remote_room_hosts):
+ """
+ Check if complexity of a remote room is too great.
+
+ Args:
+ room_id (str)
+ remote_room_hosts (list[str])
+
+ Returns: bool of whether the complexity is too great, or None
+ if unable to be fetched
+ """
+ max_complexity = self.hs.config.limit_large_room_complexity
+ complexity = yield self.federation_handler.get_room_complexity(
+ remote_room_hosts, room_id
+ )
+
+ if complexity:
+ if complexity["v1"] > max_complexity:
+ return True
+ return False
+ return None
+
+ @defer.inlineCallbacks
+ def _is_local_room_too_complex(self, room_id):
+ """
+ Check if the complexity of a local room is too great.
+
+ Args:
+ room_id (str)
+
+ Returns: bool
+ """
+ max_complexity = self.hs.config.limit_large_room_complexity
+ complexity = yield self.store.get_room_complexity(room_id)
+
+ if complexity["v1"] > max_complexity:
+ return True
+
+ return False
+
+ @defer.inlineCallbacks
def _remote_join(self, requester, remote_room_hosts, room_id, user, content):
"""Implements RoomMemberHandler._remote_join
"""
# filter ourselves out of remote_room_hosts: do_invite_join ignores it
# and if it is the only entry we'd like to return a 404 rather than a
# 500.
-
remote_room_hosts = [
host for host in remote_room_hosts if host != self.hs.hostname
]
@@ -1032,6 +1071,18 @@ class RoomMemberMasterHandler(RoomMemberHandler):
if len(remote_room_hosts) == 0:
raise SynapseError(404, "No known servers")
+ if self.hs.config.limit_large_room_joins:
+ # Fetch the room complexity
+ too_complex = yield self._is_remote_room_too_complex(
+ room_id, remote_room_hosts
+ )
+ if too_complex is True:
+ msg = "Room too large (preflight)"
+ raise SynapseError(
+ code=400, msg=msg,
+ errcode=Codes.RESOURCE_LIMIT_EXCEEDED
+ )
+
# We don't do an auth check if we are doing an invite
# join dance for now, since we're kinda implicitly checking
# that we are allowed to join when we decide whether or not we
@@ -1044,6 +1095,36 @@ class RoomMemberMasterHandler(RoomMemberHandler):
)
yield self._user_joined_room(user, room_id)
+ # Check the room we just joined wasn't too large, if we didn't fetch the
+ # complexity of it before.
+ if self.hs.config.limit_large_room_joins:
+ if too_complex is False:
+ # We checked, and we're under the limit.
+ return
+
+ # Check again, but with the local state events
+ too_complex = yield self._is_local_room_too_complex(room_id)
+
+ if too_complex is False:
+ # We're under the limit.
+ return
+
+ # The room is too large. Leave.
+ requester = types.create_requester(
+ user, None, False, None
+ )
+ yield self.update_membership(
+ requester=requester,
+ target=user,
+ room_id=room_id,
+ action="leave"
+ )
+ msg = "Room too large (postflight)"
+ raise SynapseError(
+ code=400, msg=msg,
+ errcode=Codes.RESOURCE_LIMIT_EXCEEDED
+ )
+
@defer.inlineCallbacks
def _remote_reject_invite(self, requester, remote_room_hosts, room_id, target):
"""Implements RoomMemberHandler._remote_reject_invite
diff --git a/synapse/handlers/stats.py b/synapse/handlers/stats.py
new file mode 100644
index 0000000000..0e92b405ba
--- /dev/null
+++ b/synapse/handlers/stats.py
@@ -0,0 +1,325 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018 New Vector Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from twisted.internet import defer
+
+from synapse.api.constants import EventTypes, JoinRules, Membership
+from synapse.handlers.state_deltas import StateDeltasHandler
+from synapse.metrics import event_processing_positions
+from synapse.metrics.background_process_metrics import run_as_background_process
+from synapse.types import UserID
+from synapse.util.metrics import Measure
+
+logger = logging.getLogger(__name__)
+
+
+class StatsHandler(StateDeltasHandler):
+ """Handles keeping the *_stats tables updated with a simple time-series of
+ information about the users, rooms and media on the server, such that admins
+ have some idea of who is consuming their resources.
+
+ Heavily derived from UserDirectoryHandler
+ """
+
+ def __init__(self, hs):
+ super(StatsHandler, self).__init__(hs)
+ self.hs = hs
+ self.store = hs.get_datastore()
+ self.state = hs.get_state_handler()
+ self.server_name = hs.hostname
+ self.clock = hs.get_clock()
+ self.notifier = hs.get_notifier()
+ self.is_mine_id = hs.is_mine_id
+ self.stats_bucket_size = hs.config.stats_bucket_size
+
+ # The current position in the current_state_delta stream
+ self.pos = None
+
+ # Guard to ensure we only process deltas one at a time
+ self._is_processing = False
+
+ if hs.config.stats_enabled:
+ self.notifier.add_replication_callback(self.notify_new_event)
+
+ # We kick this off so that we don't have to wait for a change before
+ # we start populating stats
+ self.clock.call_later(0, self.notify_new_event)
+
+ def notify_new_event(self):
+ """Called when there may be more deltas to process
+ """
+ if not self.hs.config.stats_enabled:
+ return
+
+ if self._is_processing:
+ return
+
+ @defer.inlineCallbacks
+ def process():
+ try:
+ yield self._unsafe_process()
+ finally:
+ self._is_processing = False
+
+ self._is_processing = True
+ run_as_background_process("stats.notify_new_event", process)
+
+ @defer.inlineCallbacks
+ def _unsafe_process(self):
+ # If self.pos is None then means we haven't fetched it from DB
+ if self.pos is None:
+ self.pos = yield self.store.get_stats_stream_pos()
+
+ # If still None then the initial background update hasn't happened yet
+ if self.pos is None:
+ defer.returnValue(None)
+
+ # Loop round handling deltas until we're up to date
+ while True:
+ with Measure(self.clock, "stats_delta"):
+ deltas = yield self.store.get_current_state_deltas(self.pos)
+ if not deltas:
+ return
+
+ logger.info("Handling %d state deltas", len(deltas))
+ yield self._handle_deltas(deltas)
+
+ self.pos = deltas[-1]["stream_id"]
+ yield self.store.update_stats_stream_pos(self.pos)
+
+ event_processing_positions.labels("stats").set(self.pos)
+
+ @defer.inlineCallbacks
+ def _handle_deltas(self, deltas):
+ """
+ Called with the state deltas to process
+ """
+ for delta in deltas:
+ typ = delta["type"]
+ state_key = delta["state_key"]
+ room_id = delta["room_id"]
+ event_id = delta["event_id"]
+ stream_id = delta["stream_id"]
+ prev_event_id = delta["prev_event_id"]
+
+ logger.debug("Handling: %r %r, %s", typ, state_key, event_id)
+
+ token = yield self.store.get_earliest_token_for_room_stats(room_id)
+
+ # If the earliest token to begin from is larger than our current
+ # stream ID, skip processing this delta.
+ if token is not None and token >= stream_id:
+ logger.debug(
+ "Ignoring: %s as earlier than this room's initial ingestion event",
+ event_id,
+ )
+ continue
+
+ if event_id is None and prev_event_id is None:
+ # Errr...
+ continue
+
+ event_content = {}
+
+ if event_id is not None:
+ event_content = (yield self.store.get_event(event_id)).content or {}
+
+ # quantise time to the nearest bucket
+ now = yield self.store.get_received_ts(event_id)
+ now = (now // 1000 // self.stats_bucket_size) * self.stats_bucket_size
+
+ if typ == EventTypes.Member:
+ # we could use _get_key_change here but it's a bit inefficient
+ # given we're not testing for a specific result; might as well
+ # just grab the prev_membership and membership strings and
+ # compare them.
+ prev_event_content = {}
+ if prev_event_id is not None:
+ prev_event_content = (
+ yield self.store.get_event(prev_event_id)
+ ).content
+
+ membership = event_content.get("membership", Membership.LEAVE)
+ prev_membership = prev_event_content.get("membership", Membership.LEAVE)
+
+ if prev_membership == membership:
+ continue
+
+ if prev_membership == Membership.JOIN:
+ yield self.store.update_stats_delta(
+ now, "room", room_id, "joined_members", -1
+ )
+ elif prev_membership == Membership.INVITE:
+ yield self.store.update_stats_delta(
+ now, "room", room_id, "invited_members", -1
+ )
+ elif prev_membership == Membership.LEAVE:
+ yield self.store.update_stats_delta(
+ now, "room", room_id, "left_members", -1
+ )
+ elif prev_membership == Membership.BAN:
+ yield self.store.update_stats_delta(
+ now, "room", room_id, "banned_members", -1
+ )
+ else:
+ err = "%s is not a valid prev_membership" % (repr(prev_membership),)
+ logger.error(err)
+ raise ValueError(err)
+
+ if membership == Membership.JOIN:
+ yield self.store.update_stats_delta(
+ now, "room", room_id, "joined_members", +1
+ )
+ elif membership == Membership.INVITE:
+ yield self.store.update_stats_delta(
+ now, "room", room_id, "invited_members", +1
+ )
+ elif membership == Membership.LEAVE:
+ yield self.store.update_stats_delta(
+ now, "room", room_id, "left_members", +1
+ )
+ elif membership == Membership.BAN:
+ yield self.store.update_stats_delta(
+ now, "room", room_id, "banned_members", +1
+ )
+ else:
+ err = "%s is not a valid membership" % (repr(membership),)
+ logger.error(err)
+ raise ValueError(err)
+
+ user_id = state_key
+ if self.is_mine_id(user_id):
+ # update user_stats as it's one of our users
+ public = yield self._is_public_room(room_id)
+
+ if membership == Membership.LEAVE:
+ yield self.store.update_stats_delta(
+ now,
+ "user",
+ user_id,
+ "public_rooms" if public else "private_rooms",
+ -1,
+ )
+ elif membership == Membership.JOIN:
+ yield self.store.update_stats_delta(
+ now,
+ "user",
+ user_id,
+ "public_rooms" if public else "private_rooms",
+ +1,
+ )
+
+ elif typ == EventTypes.Create:
+ # Newly created room. Add it with all blank portions.
+ yield self.store.update_room_state(
+ room_id,
+ {
+ "join_rules": None,
+ "history_visibility": None,
+ "encryption": None,
+ "name": None,
+ "topic": None,
+ "avatar": None,
+ "canonical_alias": None,
+ },
+ )
+
+ elif typ == EventTypes.JoinRules:
+ yield self.store.update_room_state(
+ room_id, {"join_rules": event_content.get("join_rule")}
+ )
+
+ is_public = yield self._get_key_change(
+ prev_event_id, event_id, "join_rule", JoinRules.PUBLIC
+ )
+ if is_public is not None:
+ yield self.update_public_room_stats(now, room_id, is_public)
+
+ elif typ == EventTypes.RoomHistoryVisibility:
+ yield self.store.update_room_state(
+ room_id,
+ {"history_visibility": event_content.get("history_visibility")},
+ )
+
+ is_public = yield self._get_key_change(
+ prev_event_id, event_id, "history_visibility", "world_readable"
+ )
+ if is_public is not None:
+ yield self.update_public_room_stats(now, room_id, is_public)
+
+ elif typ == EventTypes.Encryption:
+ yield self.store.update_room_state(
+ room_id, {"encryption": event_content.get("algorithm")}
+ )
+ elif typ == EventTypes.Name:
+ yield self.store.update_room_state(
+ room_id, {"name": event_content.get("name")}
+ )
+ elif typ == EventTypes.Topic:
+ yield self.store.update_room_state(
+ room_id, {"topic": event_content.get("topic")}
+ )
+ elif typ == EventTypes.RoomAvatar:
+ yield self.store.update_room_state(
+ room_id, {"avatar": event_content.get("url")}
+ )
+ elif typ == EventTypes.CanonicalAlias:
+ yield self.store.update_room_state(
+ room_id, {"canonical_alias": event_content.get("alias")}
+ )
+
+ @defer.inlineCallbacks
+ def update_public_room_stats(self, ts, room_id, is_public):
+ """
+ Increment/decrement a user's number of public rooms when a room they are
+ in changes to/from public visibility.
+
+ Args:
+ ts (int): Timestamp in seconds
+ room_id (str)
+ is_public (bool)
+ """
+ # For now, blindly iterate over all local users in the room so that
+ # we can handle the whole problem of copying buckets over as needed
+ user_ids = yield self.store.get_users_in_room(room_id)
+
+ for user_id in user_ids:
+ if self.hs.is_mine(UserID.from_string(user_id)):
+ yield self.store.update_stats_delta(
+ ts, "user", user_id, "public_rooms", +1 if is_public else -1
+ )
+ yield self.store.update_stats_delta(
+ ts, "user", user_id, "private_rooms", -1 if is_public else +1
+ )
+
+ @defer.inlineCallbacks
+ def _is_public_room(self, room_id):
+ join_rules = yield self.state.get_current_state(room_id, EventTypes.JoinRules)
+ history_visibility = yield self.state.get_current_state(
+ room_id, EventTypes.RoomHistoryVisibility
+ )
+
+ if (join_rules and join_rules.content.get("join_rule") == JoinRules.PUBLIC) or (
+ (
+ history_visibility
+ and history_visibility.content.get("history_visibility")
+ == "world_readable"
+ )
+ ):
+ defer.returnValue(True)
+ else:
+ defer.returnValue(False)
diff --git a/synapse/rest/admin/__init__.py b/synapse/rest/admin/__init__.py
index 744d85594f..d6c4dcdb18 100644
--- a/synapse/rest/admin/__init__.py
+++ b/synapse/rest/admin/__init__.py
@@ -822,10 +822,16 @@ class AdminRestResource(JsonResource):
def __init__(self, hs):
JsonResource.__init__(self, hs, canonical_json=False)
+ register_servlets(hs, self)
- register_servlets_for_client_rest_resource(hs, self)
- SendServerNoticeServlet(hs).register(self)
- VersionServlet(hs).register(self)
+
+def register_servlets(hs, http_server):
+ """
+ Register all the admin servlets.
+ """
+ register_servlets_for_client_rest_resource(hs, http_server)
+ SendServerNoticeServlet(hs).register(http_server)
+ VersionServlet(hs).register(http_server)
def register_servlets_for_client_rest_resource(hs, http_server):
diff --git a/synapse/server.py b/synapse/server.py
index 80d40b9272..9229a68a8d 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -72,6 +72,7 @@ from synapse.handlers.room_list import RoomListHandler
from synapse.handlers.room_member import RoomMemberMasterHandler
from synapse.handlers.room_member_worker import RoomMemberWorkerHandler
from synapse.handlers.set_password import SetPasswordHandler
+from synapse.handlers.stats import StatsHandler
from synapse.handlers.sync import SyncHandler
from synapse.handlers.typing import TypingHandler
from synapse.handlers.user_directory import UserDirectoryHandler
@@ -139,6 +140,7 @@ class HomeServer(object):
'acme_handler',
'auth_handler',
'device_handler',
+ 'stats_handler',
'e2e_keys_handler',
'e2e_room_keys_handler',
'event_handler',
@@ -191,6 +193,7 @@ class HomeServer(object):
REQUIRED_ON_MASTER_STARTUP = [
"user_directory_handler",
+ "stats_handler"
]
# This is overridden in derived application classes
@@ -474,6 +477,9 @@ class HomeServer(object):
def build_secrets(self):
return Secrets()
+ def build_stats_handler(self):
+ return StatsHandler(self)
+
def build_spam_checker(self):
return SpamChecker(self)
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 56c434d4e8..71316f7d09 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -56,6 +56,7 @@ from .roommember import RoomMemberStore
from .search import SearchStore
from .signatures import SignatureStore
from .state import StateStore
+from .stats import StatsStore
from .stream import StreamStore
from .tags import TagsStore
from .transactions import TransactionStore
@@ -102,6 +103,7 @@ class DataStore(
GroupServerStore,
UserErasureStore,
MonthlyActiveUsersStore,
+ StatsStore,
RelationsStore,
):
def __init__(self, db_conn, hs):
diff --git a/synapse/storage/events_worker.py b/synapse/storage/events_worker.py
index adc6cf26b5..0c569d0c6c 100644
--- a/synapse/storage/events_worker.py
+++ b/synapse/storage/events_worker.py
@@ -13,6 +13,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import division
+
import itertools
import logging
from collections import namedtuple
@@ -611,3 +613,54 @@ class EventsWorkerStore(SQLBaseStore):
return res
return self.runInteraction("get_rejection_reasons", f)
+
+ def _get_current_state_event_counts_txn(self, txn, room_id):
+ """
+ See get_current_state_event_counts.
+ """
+ sql = "SELECT COUNT(*) FROM current_state_events WHERE room_id=?"
+ txn.execute(sql, (room_id,))
+ row = txn.fetchone()
+ return row[0] if row else 0
+
+ def _get_total_state_event_counts_txn(self, txn, room_id):
+ """
+ See get_total_state_event_counts.
+ """
+ sql = "SELECT COUNT(*) FROM state_events WHERE room_id=?"
+ txn.execute(sql, (room_id,))
+ row = txn.fetchone()
+ return row[0] if row else 0
+
+ def get_current_state_event_counts(self, room_id):
+ """
+ Gets the current number of state events in a room.
+
+ Args:
+ room_id (str)
+
+ Returns:
+ Deferred[int]
+ """
+ return self.runInteraction(
+ "get_current_state_event_counts", self._get_current_state_event_counts_txn, room_id
+ )
+
+ @defer.inlineCallbacks
+ def get_room_complexity(self, room_id):
+ """
+ Get the complexity of a room.
+
+ Args:
+ room_id (str)
+
+ Returns:
+ Deferred[dict[str:int]] of complexity version to complexity.
+ """
+ state_events = yield self.get_current_state_event_counts(room_id)
+
+ # Call this one "v1", so we can introduce new ones as we want to develop
+ # it.
+ complexity_v1 = round(state_events / 500, 2)
+
+ defer.returnValue({"v1": complexity_v1})
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index 57df17bcc2..4bd1669458 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -142,6 +142,38 @@ class RoomMemberWorkerStore(EventsWorkerStore):
return self.runInteraction("get_room_summary", _get_room_summary_txn)
+ def _get_user_count_in_room_txn(self, txn, room_id, membership):
+ """
+ See get_user_count_in_room.
+ """
+ sql = (
+ "SELECT count(*) FROM room_memberships as m"
+ " INNER JOIN current_state_events as c"
+ " ON m.event_id = c.event_id "
+ " AND m.room_id = c.room_id "
+ " AND m.user_id = c.state_key"
+ " WHERE c.type = 'm.room.member' AND c.room_id = ? AND m.membership = ?"
+ )
+
+ txn.execute(sql, (room_id, membership))
+ row = txn.fetchone()
+ return row[0]
+
+ def get_user_count_in_room(self, room_id, membership):
+ """
+ Get the user count in a room with a particular membership.
+
+ Args:
+ room_id (str)
+ membership (Membership)
+
+ Returns:
+ Deferred[int]
+ """
+ return self.runInteraction(
+ "get_users_in_room", self._get_user_count_in_room_txn, room_id, membership
+ )
+
@cached()
def get_invited_rooms_for_user(self, user_id):
""" Get all the rooms the user is invited to
diff --git a/synapse/storage/schema/delta/54/stats.sql b/synapse/storage/schema/delta/54/stats.sql
new file mode 100644
index 0000000000..652e58308e
--- /dev/null
+++ b/synapse/storage/schema/delta/54/stats.sql
@@ -0,0 +1,80 @@
+/* Copyright 2018 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE TABLE stats_stream_pos (
+ Lock CHAR(1) NOT NULL DEFAULT 'X' UNIQUE, -- Makes sure this table only has one row.
+ stream_id BIGINT,
+ CHECK (Lock='X')
+);
+
+INSERT INTO stats_stream_pos (stream_id) VALUES (null);
+
+CREATE TABLE user_stats (
+ user_id TEXT NOT NULL,
+ ts BIGINT NOT NULL,
+ bucket_size INT NOT NULL,
+ public_rooms INT NOT NULL,
+ private_rooms INT NOT NULL
+);
+
+CREATE UNIQUE INDEX user_stats_user_ts ON user_stats(user_id, ts);
+
+CREATE TABLE room_stats (
+ room_id TEXT NOT NULL,
+ ts BIGINT NOT NULL,
+ bucket_size INT NOT NULL,
+ current_state_events INT NOT NULL,
+ joined_members INT NOT NULL,
+ invited_members INT NOT NULL,
+ left_members INT NOT NULL,
+ banned_members INT NOT NULL,
+ state_events INT NOT NULL
+);
+
+CREATE UNIQUE INDEX room_stats_room_ts ON room_stats(room_id, ts);
+
+-- cache of current room state; useful for the publicRooms list
+CREATE TABLE room_state (
+ room_id TEXT NOT NULL,
+ join_rules TEXT,
+ history_visibility TEXT,
+ encryption TEXT,
+ name TEXT,
+ topic TEXT,
+ avatar TEXT,
+ canonical_alias TEXT
+ -- get aliases straight from the right table
+);
+
+CREATE UNIQUE INDEX room_state_room ON room_state(room_id);
+
+CREATE TABLE room_stats_earliest_token (
+ room_id TEXT NOT NULL,
+ token BIGINT NOT NULL
+);
+
+CREATE UNIQUE INDEX room_stats_earliest_token_idx ON room_stats_earliest_token(room_id);
+
+-- Set up staging tables
+INSERT INTO background_updates (update_name, progress_json) VALUES
+ ('populate_stats_createtables', '{}');
+
+-- Run through each room and update stats
+INSERT INTO background_updates (update_name, progress_json, depends_on) VALUES
+ ('populate_stats_process_rooms', '{}', 'populate_stats_createtables');
+
+-- Clean up staging tables
+INSERT INTO background_updates (update_name, progress_json, depends_on) VALUES
+ ('populate_stats_cleanup', '{}', 'populate_stats_process_rooms');
diff --git a/synapse/storage/state_deltas.py b/synapse/storage/state_deltas.py
index 31a0279b18..5fdb442104 100644
--- a/synapse/storage/state_deltas.py
+++ b/synapse/storage/state_deltas.py
@@ -84,10 +84,16 @@ class StateDeltasStore(SQLBaseStore):
"get_current_state_deltas", get_current_state_deltas_txn
)
- def get_max_stream_id_in_current_state_deltas(self):
- return self._simple_select_one_onecol(
+ def _get_max_stream_id_in_current_state_deltas_txn(self, txn):
+ return self._simple_select_one_onecol_txn(
+ txn,
table="current_state_delta_stream",
keyvalues={},
retcol="COALESCE(MAX(stream_id), -1)",
- desc="get_max_stream_id_in_current_state_deltas",
+ )
+
+ def get_max_stream_id_in_current_state_deltas(self):
+ return self.runInteraction(
+ "get_max_stream_id_in_current_state_deltas",
+ self._get_max_stream_id_in_current_state_deltas_txn,
)
diff --git a/synapse/storage/stats.py b/synapse/storage/stats.py
new file mode 100644
index 0000000000..71b80a891d
--- /dev/null
+++ b/synapse/storage/stats.py
@@ -0,0 +1,450 @@
+# -*- coding: utf-8 -*-
+# Copyright 2018, 2019 New Vector Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from twisted.internet import defer
+
+from synapse.api.constants import EventTypes, Membership
+from synapse.storage.state_deltas import StateDeltasStore
+from synapse.util.caches.descriptors import cached
+
+logger = logging.getLogger(__name__)
+
+# these fields track absolutes (e.g. total number of rooms on the server)
+ABSOLUTE_STATS_FIELDS = {
+ "room": (
+ "current_state_events",
+ "joined_members",
+ "invited_members",
+ "left_members",
+ "banned_members",
+ "state_events",
+ ),
+ "user": ("public_rooms", "private_rooms"),
+}
+
+TYPE_TO_ROOM = {"room": ("room_stats", "room_id"), "user": ("user_stats", "user_id")}
+
+TEMP_TABLE = "_temp_populate_stats"
+
+
+class StatsStore(StateDeltasStore):
+ def __init__(self, db_conn, hs):
+ super(StatsStore, self).__init__(db_conn, hs)
+
+ self.server_name = hs.hostname
+ self.clock = self.hs.get_clock()
+ self.stats_enabled = hs.config.stats_enabled
+ self.stats_bucket_size = hs.config.stats_bucket_size
+
+ self.register_background_update_handler(
+ "populate_stats_createtables", self._populate_stats_createtables
+ )
+ self.register_background_update_handler(
+ "populate_stats_process_rooms", self._populate_stats_process_rooms
+ )
+ self.register_background_update_handler(
+ "populate_stats_cleanup", self._populate_stats_cleanup
+ )
+
+ @defer.inlineCallbacks
+ def _populate_stats_createtables(self, progress, batch_size):
+
+ if not self.stats_enabled:
+ yield self._end_background_update("populate_stats_createtables")
+ defer.returnValue(1)
+
+ # Get all the rooms that we want to process.
+ def _make_staging_area(txn):
+ sql = (
+ "CREATE TABLE IF NOT EXISTS "
+ + TEMP_TABLE
+ + "_rooms(room_id TEXT NOT NULL, events BIGINT NOT NULL)"
+ )
+ txn.execute(sql)
+
+ sql = (
+ "CREATE TABLE IF NOT EXISTS "
+ + TEMP_TABLE
+ + "_position(position TEXT NOT NULL)"
+ )
+ txn.execute(sql)
+
+ # Get rooms we want to process from the database
+ sql = """
+ SELECT room_id, count(*) FROM current_state_events
+ GROUP BY room_id
+ """
+ txn.execute(sql)
+ rooms = [{"room_id": x[0], "events": x[1]} for x in txn.fetchall()]
+ self._simple_insert_many_txn(txn, TEMP_TABLE + "_rooms", rooms)
+ del rooms
+
+ new_pos = yield self.get_max_stream_id_in_current_state_deltas()
+ yield self.runInteraction("populate_stats_temp_build", _make_staging_area)
+ yield self._simple_insert(TEMP_TABLE + "_position", {"position": new_pos})
+ self.get_earliest_token_for_room_stats.invalidate_all()
+
+ yield self._end_background_update("populate_stats_createtables")
+ defer.returnValue(1)
+
+ @defer.inlineCallbacks
+ def _populate_stats_cleanup(self, progress, batch_size):
+ """
+ Update the user directory stream position, then clean up the old tables.
+ """
+ if not self.stats_enabled:
+ yield self._end_background_update("populate_stats_cleanup")
+ defer.returnValue(1)
+
+ position = yield self._simple_select_one_onecol(
+ TEMP_TABLE + "_position", None, "position"
+ )
+ yield self.update_stats_stream_pos(position)
+
+ def _delete_staging_area(txn):
+ txn.execute("DROP TABLE IF EXISTS " + TEMP_TABLE + "_rooms")
+ txn.execute("DROP TABLE IF EXISTS " + TEMP_TABLE + "_position")
+
+ yield self.runInteraction("populate_stats_cleanup", _delete_staging_area)
+
+ yield self._end_background_update("populate_stats_cleanup")
+ defer.returnValue(1)
+
+ @defer.inlineCallbacks
+ def _populate_stats_process_rooms(self, progress, batch_size):
+
+ if not self.stats_enabled:
+ yield self._end_background_update("populate_stats_process_rooms")
+ defer.returnValue(1)
+
+ # If we don't have progress filed, delete everything.
+ if not progress:
+ yield self.delete_all_stats()
+
+ def _get_next_batch(txn):
+ # Only fetch 250 rooms, so we don't fetch too many at once, even
+ # if those 250 rooms have less than batch_size state events.
+ sql = """
+ SELECT room_id, events FROM %s_rooms
+ ORDER BY events DESC
+ LIMIT 250
+ """ % (
+ TEMP_TABLE,
+ )
+ txn.execute(sql)
+ rooms_to_work_on = txn.fetchall()
+
+ if not rooms_to_work_on:
+ return None
+
+ # Get how many are left to process, so we can give status on how
+ # far we are in processing
+ txn.execute("SELECT COUNT(*) FROM " + TEMP_TABLE + "_rooms")
+ progress["remaining"] = txn.fetchone()[0]
+
+ return rooms_to_work_on
+
+ rooms_to_work_on = yield self.runInteraction(
+ "populate_stats_temp_read", _get_next_batch
+ )
+
+ # No more rooms -- complete the transaction.
+ if not rooms_to_work_on:
+ yield self._end_background_update("populate_stats_process_rooms")
+ defer.returnValue(1)
+
+ logger.info(
+ "Processing the next %d rooms of %d remaining",
+ (len(rooms_to_work_on), progress["remaining"]),
+ )
+
+ # Number of state events we've processed by going through each room
+ processed_event_count = 0
+
+ for room_id, event_count in rooms_to_work_on:
+
+ current_state_ids = yield self.get_current_state_ids(room_id)
+
+ join_rules = yield self.get_event(
+ current_state_ids.get((EventTypes.JoinRules, "")), allow_none=True
+ )
+ history_visibility = yield self.get_event(
+ current_state_ids.get((EventTypes.RoomHistoryVisibility, "")),
+ allow_none=True,
+ )
+ encryption = yield self.get_event(
+ current_state_ids.get((EventTypes.RoomEncryption, "")), allow_none=True
+ )
+ name = yield self.get_event(
+ current_state_ids.get((EventTypes.Name, "")), allow_none=True
+ )
+ topic = yield self.get_event(
+ current_state_ids.get((EventTypes.Topic, "")), allow_none=True
+ )
+ avatar = yield self.get_event(
+ current_state_ids.get((EventTypes.RoomAvatar, "")), allow_none=True
+ )
+ canonical_alias = yield self.get_event(
+ current_state_ids.get((EventTypes.CanonicalAlias, "")), allow_none=True
+ )
+
+ def _or_none(x, arg):
+ if x:
+ return x.content.get(arg)
+ return None
+
+ yield self.update_room_state(
+ room_id,
+ {
+ "join_rules": _or_none(join_rules, "join_rule"),
+ "history_visibility": _or_none(
+ history_visibility, "history_visibility"
+ ),
+ "encryption": _or_none(encryption, "algorithm"),
+ "name": _or_none(name, "name"),
+ "topic": _or_none(topic, "topic"),
+ "avatar": _or_none(avatar, "url"),
+ "canonical_alias": _or_none(canonical_alias, "alias"),
+ },
+ )
+
+ now = self.hs.get_reactor().seconds()
+
+ # quantise time to the nearest bucket
+ now = (now // self.stats_bucket_size) * self.stats_bucket_size
+
+ def _fetch_data(txn):
+
+ # Get the current token of the room
+ current_token = self._get_max_stream_id_in_current_state_deltas_txn(txn)
+
+ current_state_events = len(current_state_ids)
+ joined_members = self._get_user_count_in_room_txn(
+ txn, room_id, Membership.JOIN
+ )
+ invited_members = self._get_user_count_in_room_txn(
+ txn, room_id, Membership.INVITE
+ )
+ left_members = self._get_user_count_in_room_txn(
+ txn, room_id, Membership.LEAVE
+ )
+ banned_members = self._get_user_count_in_room_txn(
+ txn, room_id, Membership.BAN
+ )
+ total_state_events = self._get_total_state_event_counts_txn(
+ txn, room_id
+ )
+
+ self._update_stats_txn(
+ txn,
+ "room",
+ room_id,
+ now,
+ {
+ "bucket_size": self.stats_bucket_size,
+ "current_state_events": current_state_events,
+ "joined_members": joined_members,
+ "invited_members": invited_members,
+ "left_members": left_members,
+ "banned_members": banned_members,
+ "state_events": total_state_events,
+ },
+ )
+ self._simple_insert_txn(
+ txn,
+ "room_stats_earliest_token",
+ {"room_id": room_id, "token": current_token},
+ )
+
+ yield self.runInteraction("update_room_stats", _fetch_data)
+
+ # We've finished a room. Delete it from the table.
+ yield self._simple_delete_one(TEMP_TABLE + "_rooms", {"room_id": room_id})
+ # Update the remaining counter.
+ progress["remaining"] -= 1
+ yield self.runInteraction(
+ "populate_stats",
+ self._background_update_progress_txn,
+ "populate_stats_process_rooms",
+ progress,
+ )
+
+ processed_event_count += event_count
+
+ if processed_event_count > batch_size:
+ # Don't process any more rooms, we've hit our batch size.
+ defer.returnValue(processed_event_count)
+
+ defer.returnValue(processed_event_count)
+
+ def delete_all_stats(self):
+ """
+ Delete all statistics records.
+ """
+
+ def _delete_all_stats_txn(txn):
+ txn.execute("DELETE FROM room_state")
+ txn.execute("DELETE FROM room_stats")
+ txn.execute("DELETE FROM room_stats_earliest_token")
+ txn.execute("DELETE FROM user_stats")
+
+ return self.runInteraction("delete_all_stats", _delete_all_stats_txn)
+
+ def get_stats_stream_pos(self):
+ return self._simple_select_one_onecol(
+ table="stats_stream_pos",
+ keyvalues={},
+ retcol="stream_id",
+ desc="stats_stream_pos",
+ )
+
+ def update_stats_stream_pos(self, stream_id):
+ return self._simple_update_one(
+ table="stats_stream_pos",
+ keyvalues={},
+ updatevalues={"stream_id": stream_id},
+ desc="update_stats_stream_pos",
+ )
+
+ def update_room_state(self, room_id, fields):
+ """
+ Args:
+ room_id (str)
+ fields (dict[str:Any])
+ """
+ return self._simple_upsert(
+ table="room_state",
+ keyvalues={"room_id": room_id},
+ values=fields,
+ desc="update_room_state",
+ )
+
+ def get_deltas_for_room(self, room_id, start, size=100):
+ """
+ Get statistics deltas for a given room.
+
+ Args:
+ room_id (str)
+ start (int): Pagination start. Number of entries, not timestamp.
+ size (int): How many entries to return.
+
+ Returns:
+ Deferred[list[dict]], where the dict has the keys of
+ ABSOLUTE_STATS_FIELDS["room"] and "ts".
+ """
+ return self._simple_select_list_paginate(
+ "room_stats",
+ {"room_id": room_id},
+ "ts",
+ start,
+ size,
+ retcols=(list(ABSOLUTE_STATS_FIELDS["room"]) + ["ts"]),
+ order_direction="DESC",
+ )
+
+ def get_all_room_state(self):
+ return self._simple_select_list(
+ "room_state", None, retcols=("name", "topic", "canonical_alias")
+ )
+
+ @cached()
+ def get_earliest_token_for_room_stats(self, room_id):
+ """
+ Fetch the "earliest token". This is used by the room stats delta
+ processor to ignore deltas that have been processed between the
+ start of the background task and any particular room's stats
+ being calculated.
+
+ Returns:
+ Deferred[int]
+ """
+ return self._simple_select_one_onecol(
+ "room_stats_earliest_token",
+ {"room_id": room_id},
+ retcol="token",
+ allow_none=True,
+ )
+
+ def update_stats(self, stats_type, stats_id, ts, fields):
+ table, id_col = TYPE_TO_ROOM[stats_type]
+ return self._simple_upsert(
+ table=table,
+ keyvalues={id_col: stats_id, "ts": ts},
+ values=fields,
+ desc="update_stats",
+ )
+
+ def _update_stats_txn(self, txn, stats_type, stats_id, ts, fields):
+ table, id_col = TYPE_TO_ROOM[stats_type]
+ return self._simple_upsert_txn(
+ txn, table=table, keyvalues={id_col: stats_id, "ts": ts}, values=fields
+ )
+
+ def update_stats_delta(self, ts, stats_type, stats_id, field, value):
+ def _update_stats_delta(txn):
+ table, id_col = TYPE_TO_ROOM[stats_type]
+
+ sql = (
+ "SELECT * FROM %s"
+ " WHERE %s=? and ts=("
+ " SELECT MAX(ts) FROM %s"
+ " WHERE %s=?"
+ ")"
+ ) % (table, id_col, table, id_col)
+ txn.execute(sql, (stats_id, stats_id))
+ rows = self.cursor_to_dict(txn)
+ if len(rows) == 0:
+ # silently skip as we don't have anything to apply a delta to yet.
+ # this tries to minimise any race between the initial sync and
+ # subsequent deltas arriving.
+ return
+
+ current_ts = ts
+ latest_ts = rows[0]["ts"]
+ if current_ts < latest_ts:
+ # This one is in the past, but we're just encountering it now.
+ # Mark it as part of the current bucket.
+ current_ts = latest_ts
+ elif ts != latest_ts:
+ # we have to copy our absolute counters over to the new entry.
+ values = {
+ key: rows[0][key] for key in ABSOLUTE_STATS_FIELDS[stats_type]
+ }
+ values[id_col] = stats_id
+ values["ts"] = ts
+ values["bucket_size"] = self.stats_bucket_size
+
+ self._simple_insert_txn(txn, table=table, values=values)
+
+ # actually update the new value
+ if stats_type in ABSOLUTE_STATS_FIELDS[stats_type]:
+ self._simple_update_txn(
+ txn,
+ table=table,
+ keyvalues={id_col: stats_id, "ts": current_ts},
+ updatevalues={field: value},
+ )
+ else:
+ sql = ("UPDATE %s SET %s=%s+? WHERE %s=? AND ts=?") % (
+ table,
+ field,
+ field,
+ id_col,
+ )
+ txn.execute(sql, (value, stats_id, current_ts))
+
+ return self.runInteraction("update_stats_delta", _update_stats_delta)
diff --git a/tests/federation/test_complexity.py b/tests/federation/test_complexity.py
new file mode 100644
index 0000000000..c499d9d9d6
--- /dev/null
+++ b/tests/federation/test_complexity.py
@@ -0,0 +1,162 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 Matrix.org Foundation
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from mock import Mock
+
+from twisted.internet import defer
+
+from synapse.api.errors import Codes, SynapseError
+from synapse.config.ratelimiting import FederationRateLimitConfig
+from synapse.federation.transport import server
+from synapse.rest import admin
+from synapse.rest.client.v1 import login, room
+from synapse.types import UserID
+from synapse.util.ratelimitutils import FederationRateLimiter
+
+from tests import unittest
+
+
+class RoomComplexityTests(unittest.HomeserverTestCase):
+
+ servlets = [
+ admin.register_servlets,
+ room.register_servlets,
+ login.register_servlets,
+ ]
+
+ def default_config(self, name='test'):
+ config = super().default_config(name=name)
+ config["limit_large_remote_room_joins"] = True
+ config["limit_large_remote_room_complexity"] = 0.05
+ return config
+
+ def prepare(self, reactor, clock, homeserver):
+ class Authenticator(object):
+ def authenticate_request(self, request, content):
+ return defer.succeed("otherserver.nottld")
+
+ ratelimiter = FederationRateLimiter(
+ clock,
+ FederationRateLimitConfig(
+ window_size=1,
+ sleep_limit=1,
+ sleep_msec=1,
+ reject_limit=1000,
+ concurrent_requests=1000,
+ ),
+ )
+ server.register_servlets(
+ homeserver, self.resource, Authenticator(), ratelimiter
+ )
+
+ def test_complexity_simple(self):
+
+ u1 = self.register_user("u1", "pass")
+ u1_token = self.login("u1", "pass")
+
+ room_1 = self.helper.create_room_as(u1, tok=u1_token)
+ self.helper.send_state(
+ room_1, event_type="m.room.topic", body={"topic": "foo"}, tok=u1_token
+ )
+
+ # Get the room complexity
+ request, channel = self.make_request(
+ "GET", "/_matrix/federation/unstable/rooms/%s/complexity" % (room_1,)
+ )
+ self.render(request)
+ self.assertEquals(200, channel.code)
+ complexity = channel.json_body["v1"]
+ self.assertTrue(complexity > 0, complexity)
+
+ # Artificially raise the complexity
+ store = self.hs.get_datastore()
+ store.get_current_state_event_counts = lambda x: defer.succeed(500 * 1.23)
+
+ # Get the room complexity again -- make sure it's our artificial value
+ request, channel = self.make_request(
+ "GET", "/_matrix/federation/unstable/rooms/%s/complexity" % (room_1,)
+ )
+ self.render(request)
+ self.assertEquals(200, channel.code)
+ complexity = channel.json_body["v1"]
+ self.assertEqual(complexity, 1.23)
+
+ def test_join_too_large(self):
+
+ u1 = self.register_user("u1", "pass")
+
+ handler = self.hs.get_room_member_handler()
+ fed_transport = self.hs.get_federation_transport_client()
+
+ # Mock out some things, because we don't want to test the whole join
+ fed_transport.client.get_json = Mock(return_value=defer.succeed({"v1": 9999}))
+ handler.federation_handler.do_invite_join = Mock(return_value=defer.succeed(1))
+
+ d = handler._remote_join(
+ None,
+ ["otherserver.example"],
+ "roomid",
+ UserID.from_string(u1),
+ {"membership": "join"},
+ )
+
+ self.pump()
+
+ # The request failed with a SynapseError saying the resource limit was
+ # exceeded.
+ f = self.get_failure(d, SynapseError)
+ self.assertEqual(f.value.code, 400, f.value)
+ self.assertEqual(f.value.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
+
+ def test_join_too_large_once_joined(self):
+
+ u1 = self.register_user("u1", "pass")
+ u1_token = self.login("u1", "pass")
+
+ # Ok, this might seem a bit weird -- I want to test that we actually
+ # leave the room, but I don't want to simulate two servers. So, we make
+ # a local room, which we say we're joining remotely, even if there's no
+ # remote, because we mock that out. Then, we'll leave the (actually
+ # local) room, which will be propagated over federation in a real
+ # scenario.
+ room_1 = self.helper.create_room_as(u1, tok=u1_token)
+
+ handler = self.hs.get_room_member_handler()
+ fed_transport = self.hs.get_federation_transport_client()
+
+ # Mock out some things, because we don't want to test the whole join
+ fed_transport.client.get_json = Mock(return_value=defer.succeed(None))
+ handler.federation_handler.do_invite_join = Mock(return_value=defer.succeed(1))
+
+ # Artificially raise the complexity
+ self.hs.get_datastore().get_current_state_event_counts = lambda x: defer.succeed(
+ 600
+ )
+
+ d = handler._remote_join(
+ None,
+ ["otherserver.example"],
+ room_1,
+ UserID.from_string(u1),
+ {"membership": "join"},
+ )
+
+ self.pump()
+
+ # The request failed with a SynapseError saying the resource limit was
+ # exceeded.
+ f = self.get_failure(d, SynapseError)
+ self.assertEqual(f.value.code, 400)
+ self.assertEqual(f.value.errcode, Codes.RESOURCE_LIMIT_EXCEEDED)
diff --git a/tests/handlers/test_stats.py b/tests/handlers/test_stats.py
new file mode 100644
index 0000000000..249aba3d59
--- /dev/null
+++ b/tests/handlers/test_stats.py
@@ -0,0 +1,251 @@
+# -*- coding: utf-8 -*-
+# Copyright 2019 New Vector Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from mock import Mock
+
+from twisted.internet import defer
+
+from synapse.api.constants import EventTypes, Membership
+from synapse.rest import admin
+from synapse.rest.client.v1 import login, room
+
+from tests import unittest
+
+
+class StatsRoomTests(unittest.HomeserverTestCase):
+
+ servlets = [
+ admin.register_servlets_for_client_rest_resource,
+ room.register_servlets,
+ login.register_servlets,
+ ]
+
+ def prepare(self, reactor, clock, hs):
+
+ self.store = hs.get_datastore()
+ self.handler = self.hs.get_stats_handler()
+
+ def _add_background_updates(self):
+ """
+ Add the background updates we need to run.
+ """
+ # Ugh, have to reset this flag
+ self.store._all_done = False
+
+ self.get_success(
+ self.store._simple_insert(
+ "background_updates",
+ {"update_name": "populate_stats_createtables", "progress_json": "{}"},
+ )
+ )
+ self.get_success(
+ self.store._simple_insert(
+ "background_updates",
+ {
+ "update_name": "populate_stats_process_rooms",
+ "progress_json": "{}",
+ "depends_on": "populate_stats_createtables",
+ },
+ )
+ )
+ self.get_success(
+ self.store._simple_insert(
+ "background_updates",
+ {
+ "update_name": "populate_stats_cleanup",
+ "progress_json": "{}",
+ "depends_on": "populate_stats_process_rooms",
+ },
+ )
+ )
+
+ def test_initial_room(self):
+ """
+ The background updates will build the table from scratch.
+ """
+ r = self.get_success(self.store.get_all_room_state())
+ self.assertEqual(len(r), 0)
+
+ # Disable stats
+ self.hs.config.stats_enabled = False
+ self.handler.stats_enabled = False
+
+ u1 = self.register_user("u1", "pass")
+ u1_token = self.login("u1", "pass")
+
+ room_1 = self.helper.create_room_as(u1, tok=u1_token)
+ self.helper.send_state(
+ room_1, event_type="m.room.topic", body={"topic": "foo"}, tok=u1_token
+ )
+
+ # Stats disabled, shouldn't have done anything
+ r = self.get_success(self.store.get_all_room_state())
+ self.assertEqual(len(r), 0)
+
+ # Enable stats
+ self.hs.config.stats_enabled = True
+ self.handler.stats_enabled = True
+
+ # Do the initial population of the user directory via the background update
+ self._add_background_updates()
+
+ while not self.get_success(self.store.has_completed_background_updates()):
+ self.get_success(self.store.do_next_background_update(100), by=0.1)
+
+ r = self.get_success(self.store.get_all_room_state())
+
+ self.assertEqual(len(r), 1)
+ self.assertEqual(r[0]["topic"], "foo")
+
+ def test_initial_earliest_token(self):
+ """
+ Ingestion via notify_new_event will ignore tokens that the background
+ update have already processed.
+ """
+ self.reactor.advance(86401)
+
+ self.hs.config.stats_enabled = False
+ self.handler.stats_enabled = False
+
+ u1 = self.register_user("u1", "pass")
+ u1_token = self.login("u1", "pass")
+
+ u2 = self.register_user("u2", "pass")
+ u2_token = self.login("u2", "pass")
+
+ u3 = self.register_user("u3", "pass")
+ u3_token = self.login("u3", "pass")
+
+ room_1 = self.helper.create_room_as(u1, tok=u1_token)
+ self.helper.send_state(
+ room_1, event_type="m.room.topic", body={"topic": "foo"}, tok=u1_token
+ )
+
+ # Begin the ingestion by creating the temp tables. This will also store
+ # the position that the deltas should begin at, once they take over.
+ self.hs.config.stats_enabled = True
+ self.handler.stats_enabled = True
+ self.store._all_done = False
+ self.get_success(self.store.update_stats_stream_pos(None))
+
+ self.get_success(
+ self.store._simple_insert(
+ "background_updates",
+ {"update_name": "populate_stats_createtables", "progress_json": "{}"},
+ )
+ )
+
+ while not self.get_success(self.store.has_completed_background_updates()):
+ self.get_success(self.store.do_next_background_update(100), by=0.1)
+
+ # Now, before the table is actually ingested, add some more events.
+ self.helper.invite(room=room_1, src=u1, targ=u2, tok=u1_token)
+ self.helper.join(room=room_1, user=u2, tok=u2_token)
+
+ # Now do the initial ingestion.
+ self.get_success(
+ self.store._simple_insert(
+ "background_updates",
+ {"update_name": "populate_stats_process_rooms", "progress_json": "{}"},
+ )
+ )
+ self.get_success(
+ self.store._simple_insert(
+ "background_updates",
+ {
+ "update_name": "populate_stats_cleanup",
+ "progress_json": "{}",
+ "depends_on": "populate_stats_process_rooms",
+ },
+ )
+ )
+
+ self.store._all_done = False
+ while not self.get_success(self.store.has_completed_background_updates()):
+ self.get_success(self.store.do_next_background_update(100), by=0.1)
+
+ self.reactor.advance(86401)
+
+ # Now add some more events, triggering ingestion. Because of the stream
+ # position being set to before the events sent in the middle, a simpler
+ # implementation would reprocess those events, and say there were four
+ # users, not three.
+ self.helper.invite(room=room_1, src=u1, targ=u3, tok=u1_token)
+ self.helper.join(room=room_1, user=u3, tok=u3_token)
+
+ # Get the deltas! There should be two -- day 1, and day 2.
+ r = self.get_success(self.store.get_deltas_for_room(room_1, 0))
+
+ # The oldest has 2 joined members
+ self.assertEqual(r[-1]["joined_members"], 2)
+
+ # The newest has 3
+ self.assertEqual(r[0]["joined_members"], 3)
+
+ def test_incorrect_state_transition(self):
+ """
+ If the state transition is not one of (JOIN, INVITE, LEAVE, BAN) to
+ (JOIN, INVITE, LEAVE, BAN), an error is raised.
+ """
+ events = {
+ "a1": {"membership": Membership.LEAVE},
+ "a2": {"membership": "not a real thing"},
+ }
+
+ def get_event(event_id):
+ m = Mock()
+ m.content = events[event_id]
+ d = defer.Deferred()
+ self.reactor.callLater(0.0, d.callback, m)
+ return d
+
+ def get_received_ts(event_id):
+ return defer.succeed(1)
+
+ self.store.get_received_ts = get_received_ts
+ self.store.get_event = get_event
+
+ deltas = [
+ {
+ "type": EventTypes.Member,
+ "state_key": "some_user",
+ "room_id": "room",
+ "event_id": "a1",
+ "prev_event_id": "a2",
+ "stream_id": "bleb",
+ }
+ ]
+
+ f = self.get_failure(self.handler._handle_deltas(deltas), ValueError)
+ self.assertEqual(
+ f.value.args[0], "'not a real thing' is not a valid prev_membership"
+ )
+
+ # And the other way...
+ deltas = [
+ {
+ "type": EventTypes.Member,
+ "state_key": "some_user",
+ "room_id": "room",
+ "event_id": "a2",
+ "prev_event_id": "a1",
+ "stream_id": "bleb",
+ }
+ ]
+
+ f = self.get_failure(self.handler._handle_deltas(deltas), ValueError)
+ self.assertEqual(
+ f.value.args[0], "'not a real thing' is not a valid membership"
+ )
diff --git a/tests/rest/client/v1/utils.py b/tests/rest/client/v1/utils.py
index 05b0143c42..f7133fc12e 100644
--- a/tests/rest/client/v1/utils.py
+++ b/tests/rest/client/v1/utils.py
@@ -127,3 +127,20 @@ class RestHelper(object):
)
return channel.json_body
+
+ def send_state(self, room_id, event_type, body, tok, expect_code=200):
+ path = "/_matrix/client/r0/rooms/%s/state/%s" % (room_id, event_type)
+ if tok:
+ path = path + "?access_token=%s" % tok
+
+ request, channel = make_request(
+ self.hs.get_reactor(), "PUT", path, json.dumps(body).encode('utf8')
+ )
+ render(request, self.resource, self.hs.get_reactor())
+
+ assert int(channel.result["code"]) == expect_code, (
+ "Expected: %d, got: %d, resp: %r"
+ % (expect_code, int(channel.result["code"]), channel.result["body"])
+ )
+
+ return channel.json_body
diff --git a/tests/unittest.py b/tests/unittest.py
index 26204470b1..ea2b7f272e 100644
--- a/tests/unittest.py
+++ b/tests/unittest.py
@@ -22,8 +22,6 @@ from mock import Mock
from canonicaljson import json
-import twisted
-import twisted.logger
from twisted.internet.defer import Deferred
from twisted.trial import unittest
@@ -77,10 +75,6 @@ class TestCase(unittest.TestCase):
@around(self)
def setUp(orig):
- # enable debugging of delayed calls - this means that we get a
- # traceback when a unit test exits leaving things on the reactor.
- twisted.internet.base.DelayedCall.debug = True
-
# if we're not starting in the sentinel logcontext, then to be honest
# all future bets are off.
if LoggingContext.current_context() is not LoggingContext.sentinel:
|
