Require SQLite3 version 3.15 or above

This is primarily to allow tuple comparisons in queries, though a better
query optimiser and other improvements mean that using newer versions of
sqlite is highly recommended anyway.

.dockerignore

@@ -1,7 +0,0 @@
-Dockerfile
-.travis.yml
-.gitignore
-demo/etc
-env
-tox.ini
-.git

.gitignore

@@ -32,7 +32,6 @@ demo/media_store.*
 demo/etc
 
 uploads
-cache
 
 .idea/
 media_store/
@@ -49,4 +48,3 @@ env/
 *.config
 
 .vscode/
-.ropeproject/

AUTHORS.rst

@@ -60,6 +60,3 @@ Niklas Riekenbrauck <nikriek at gmail dot.com>
 
 Christoph Witzany <christoph at web.crofting.com>
  * Add LDAP support for authentication
-
-Pierre Jaury <pierre at jaury.eu>
-* Docker packaging

CHANGES.rst

@@ -1,19 +1,5 @@
-Changes in synapse v0.29.1 (2018-05-17)
-==========================================
-Changes:
-
-* Update docker documentation (PR #3222)
-
-Changes in synapse v0.29.0 (2018-05-16)
-===========================================
-Not changes since v0.29.0-rc1
-
-Changes in synapse v0.29.0-rc1 (2018-05-14)
-===========================================
-
-Notable changes, a docker file for running Synapse (Thanks to @kaiyou!) and a
-closed spec bug in the Client Server API. Additionally further prep for Python 3
-migration.
+Changes in synapse <unreleased>
+===============================
 
 Potentially breaking change:
 
@@ -26,66 +12,6 @@ Potentially breaking change:
 
   Thanks to @NotAFile for fixing this.
 
-Features:
-
-* Add a Dockerfile for synapse (PR #2846) Thanks to @kaiyou!
-
-Changes - General:
-
-* nuke-room-from-db.sh: added postgresql option and help (PR #2337) Thanks to @rubo77!
-* Part user from rooms on account deactivate (PR #3201)
-* Make 'unexpected logging context' into warnings (PR #3007)
-* Set Server header in SynapseRequest (PR #3208)
-* remove duplicates from groups tables (PR #3129)
-* Improve exception handling for background processes (PR #3138)
-* Add missing consumeErrors to improve exception handling (PR #3139)
-* reraise exceptions more carefully (PR #3142)
-* Remove redundant call to preserve_fn (PR #3143)
-* Trap exceptions thrown within run_in_background (PR #3144)
-
-Changes - Refactors:
-
-* Refactor /context to reuse pagination storage functions (PR #3193)
-* Refactor recent events func to use pagination func (PR #3195)
-* Refactor pagination DB API to return concrete type (PR #3196)
-* Refactor get_recent_events_for_room return type (PR #3198)
-* Refactor sync APIs to reuse pagination API (PR #3199)
-* Remove unused code path from member change DB func (PR #3200)
-* Refactor request handling wrappers (PR #3203)
-* transaction_id, destination defined twice (PR #3209) Thanks to @damir-manapov!
-* Refactor event storage to prepare for changes in state calculations (PR #3141)
-* Set Server header in SynapseRequest (PR #3208)
-* Use deferred.addTimeout instead of time_bound_deferred (PR #3127, #3178)
-* Use run_in_background in preference to preserve_fn (PR #3140)
-
-Changes - Python 3 migration:
-
-* Construct HMAC as bytes on py3 (PR #3156) Thanks to @NotAFile!
-* run config tests on py3 (PR #3159) Thanks to @NotAFile!
-* Open certificate files as bytes (PR #3084) Thanks to @NotAFile!
-* Open config file in non-bytes mode (PR #3085) Thanks to @NotAFile!
-* Make event properties raise AttributeError instead (PR #3102) Thanks to @NotAFile!
-* Use six.moves.urlparse (PR #3108) Thanks to @NotAFile!
-* Add py3 tests to tox with folders that work (PR #3145) Thanks to @NotAFile!
-* Don't yield in list comprehensions (PR #3150) Thanks to @NotAFile!
-* Move more xrange to six (PR #3151) Thanks to @NotAFile!
-* make imports local (PR #3152) Thanks to @NotAFile!
-* move httplib import to six (PR #3153) Thanks to @NotAFile!
-* Replace stringIO imports with six (PR #3154, #3168) Thanks to @NotAFile!
-* more bytes strings (PR #3155) Thanks to @NotAFile!
-
-Bug Fixes:
-
-* synapse fails to start under Twisted >= 18.4 (PR #3157)
-* Fix a class of logcontext leaks (PR #3170)
-* Fix a couple of logcontext leaks in unit tests (PR #3172)
-* Fix logcontext leak in media repo (PR #3174)
-* Escape label values in prometheus metrics (PR #3175, #3186)
-* Fix 'Unhandled Error' logs with Twisted 18.4 (PR #3182) Thanks to @Half-Shot!
-* Fix logcontext leaks in rate limiter (PR #3183)
-* notifications: Convert next_token to string according to the spec (PR #3190) Thanks to @mujx!
-* nuke-room-from-db.sh: fix deletion from search table (PR #3194) Thanks to @rubo77!
-* add guard for None on purge_history api (PR #3160) Thanks to @krombel!
 
 Changes in synapse v0.28.1 (2018-05-01)
 =======================================

Dockerfile

@@ -1,19 +0,0 @@
-FROM docker.io/python:2-alpine3.7
-
-RUN apk add --no-cache --virtual .nacl_deps su-exec build-base libffi-dev zlib-dev libressl-dev libjpeg-turbo-dev linux-headers postgresql-dev
-
-COPY . /synapse
-
-# A wheel cache may be provided in ./cache for faster build
-RUN cd /synapse \
- && pip install --upgrade pip setuptools psycopg2 \
- && mkdir -p /synapse/cache \
- && pip install -f /synapse/cache --upgrade --process-dependency-links . \
- && mv /synapse/contrib/docker/start.py /synapse/contrib/docker/conf / \
- && rm -rf setup.py setup.cfg synapse
-
-VOLUME ["/data"]
-
-EXPOSE 8008/tcp 8448/tcp
-
-ENTRYPOINT ["/start.py"]

MANIFEST.in

@@ -25,8 +25,6 @@ recursive-include synapse/static *.js
 exclude jenkins.sh
 exclude jenkins*.sh
 exclude jenkins*
-exclude Dockerfile
-exclude .dockerignore
 recursive-exclude jenkins *.sh
 
 prune .github

README.rst

@@ -157,9 +157,8 @@ if you prefer.
 
 In case of problems, please see the _`Troubleshooting` section below.
 
-There is an offical synapse image available at https://hub.docker.com/r/matrixdotorg/synapse/tags/ which can be used with the docker-compose file available at `contrib/docker`. Further information on this including configuration options is available in `contrib/docker/README.md`.
-
-Alternatively, Andreas Peters (previously Silvio Fricke) has contributed a Dockerfile to automate a synapse server in a single Docker image, at https://hub.docker.com/r/avhost/docker-matrix/tags/
+Alternatively, Andreas Peters (previously Silvio Fricke) has contributed a Dockerfile to automate the
+above in Docker at https://hub.docker.com/r/avhost/docker-matrix/tags/
 
 Also, Martin Giess has created an auto-deployment process with vagrant/ansible,
 tested with VirtualBox/AWS/DigitalOcean - see https://github.com/EMnify/matrix-synapse-auto-deploy

contrib/docker/README.md

@@ -1,153 +0,0 @@
-# Synapse Docker
-
-The `matrixdotorg/synapse` Docker image will run Synapse as a single process. It does not provide a
-database server or a TURN server, you should run these separately.
-
-If you run a Postgres server, you should simply include it in the same Compose
-project or set the proper environment variables and the image will automatically
-use that server.
-
-## Build
-
-Build the docker image with the `docker build` command from the root of the synapse repository.
-
-```
-docker build -t docker.io/matrixdotorg/synapse .
-```
-
-The `-t` option sets the image tag. Official images are tagged `matrixdotorg/synapse:<version>` where `<version>` is the same as the release tag in the synapse git repository.
-
-You may have a local Python wheel cache available, in which case copy the relevant packages in the ``cache/`` directory at the root of the project.
-
-## Run
-
-This image is designed to run either with an automatically generated configuration
-file or with a custom configuration that requires manual edition.
-
-### Automated configuration
-
-It is recommended that you use Docker Compose to run your containers, including
-this image and a Postgres server. A sample ``docker-compose.yml`` is provided,
-including example labels for reverse proxying and other artifacts.
-
-Read the section about environment variables and set at least mandatory variables,
-then run the server:
-
-```
-docker-compose up -d
-```
-
-If secrets are not specified in the environment variables, they will be generated
-as part of the startup. Please ensure these secrets are kept between launches of the
-Docker container, as their loss may require users to log in again.
-
-### Manual configuration
-
-A sample ``docker-compose.yml`` is provided, including example labels for
-reverse proxying and other artifacts. The docker-compose file is an example,
-please comment/uncomment sections that are not suitable for your usecase.
-
-Specify a ``SYNAPSE_CONFIG_PATH``, preferably to a persistent path,
-to use manual configuration. To generate a fresh ``homeserver.yaml``, simply run:
-
-```
-docker-compose run --rm -e SYNAPSE_SERVER_NAME=my.matrix.host synapse generate
-```
-
-Then, customize your configuration and run the server:
-
-```
-docker-compose up -d
-```
-
-### Without Compose
-
-If you do not wish to use Compose, you may still run this image using plain
-Docker commands. Note that the following is just a guideline and you may need
-to add parameters to the docker run command to account for the network situation
-with your postgres database.
-
-```
-docker run \
-    -d \
-    --name synapse \
-    -v ${DATA_PATH}:/data \
-    -e SYNAPSE_SERVER_NAME=my.matrix.host \
-    -e SYNAPSE_REPORT_STATS=yes \
-    docker.io/matrixdotorg/synapse:latest
-```
-
-## Volumes
-
-The image expects a single volume, located at ``/data``, that will hold:
-
-* temporary files during uploads;
-* uploaded media and thumbnails;
-* the SQLite database if you do not configure postgres;
-* the appservices configuration.
-
-You are free to use separate volumes depending on storage endpoints at your
-disposal. For instance, ``/data/media`` coud be stored on a large but low
-performance hdd storage while other files could be stored on high performance
-endpoints.
-
-In order to setup an application service, simply create an ``appservices``
-directory in the data volume and write the application service Yaml
-configuration file there. Multiple application services are supported.
-
-## Environment
-
-Unless you specify a custom path for the configuration file, a very generic
-file will be generated, based on the following environment settings.
-These are a good starting point for setting up your own deployment.
-
-Global settings:
-
-* ``UID``, the user id Synapse will run as [default 991]
-* ``GID``, the group id Synapse will run as [default 991]
-* ``SYNAPSE_CONFIG_PATH``, path to a custom config file
-
-If ``SYNAPSE_CONFIG_PATH`` is set, you should generate a configuration file
-then customize it manually. No other environment variable is required.
-
-Otherwise, a dynamic configuration file will be used. The following environment
-variables are available for configuration:
-
-* ``SYNAPSE_SERVER_NAME`` (mandatory), the current server public hostname.
-* ``SYNAPSE_REPORT_STATS``, (mandatory, ``yes`` or ``no``), enable anonymous
-  statistics reporting back to the Matrix project which helps us to get funding.
-* ``SYNAPSE_NO_TLS``, set this variable to disable TLS in Synapse (use this if
-  you run your own TLS-capable reverse proxy).
-* ``SYNAPSE_ENABLE_REGISTRATION``, set this variable to enable registration on
-  the Synapse instance.
-* ``SYNAPSE_ALLOW_GUEST``, set this variable to allow guest joining this server.
-* ``SYNAPSE_EVENT_CACHE_SIZE``, the event cache size [default `10K`].
-* ``SYNAPSE_CACHE_FACTOR``, the cache factor [default `0.5`].
-* ``SYNAPSE_RECAPTCHA_PUBLIC_KEY``, set this variable to the recaptcha public
-  key in order to enable recaptcha upon registration.
-* ``SYNAPSE_RECAPTCHA_PRIVATE_KEY``, set this variable to the recaptcha private
-  key in order to enable recaptcha upon registration.
-* ``SYNAPSE_TURN_URIS``, set this variable to the coma-separated list of TURN
-  uris to enable TURN for this homeserver.
-* ``SYNAPSE_TURN_SECRET``, set this to the TURN shared secret if required.
-
-Shared secrets, that will be initialized to random values if not set:
-
-* ``SYNAPSE_REGISTRATION_SHARED_SECRET``, secret for registrering users if
-  registration is disable.
-* ``SYNAPSE_MACAROON_SECRET_KEY`` secret for signing access tokens
-  to the server.
-
-Database specific values (will use SQLite if not set):
-
-* `POSTGRES_DB` - The database name for the synapse postgres database. [default: `synapse`]
-* `POSTGRES_HOST` - The host of the postgres database if you wish to use postgresql instead of sqlite3. [default: `db` which is useful when using a container on the same docker network in a compose file where the postgres service is called `db`]
-* `POSTGRES_PASSWORD` - The password for the synapse postgres database. **If this is set then postgres will be used instead of sqlite3.** [default: none] **NOTE**: You are highly encouraged to use postgresql! Please use the compose file to make it easier to deploy.
-* `POSTGRES_USER` - The user for the synapse postgres database. [default: `matrix`]
-
-Mail server specific values (will not send emails if not set):
-
-* ``SYNAPSE_SMTP_HOST``, hostname to the mail server.
-* ``SYNAPSE_SMTP_PORT``, TCP port for accessing the mail server [default ``25``].
-* ``SYNAPSE_SMTP_USER``, username for authenticating against the mail server if any.
-* ``SYNAPSE_SMTP_PASSWORD``, password for authenticating against the mail server if any.

contrib/docker/conf/homeserver.yaml

@@ -1,219 +0,0 @@
-# vim:ft=yaml
-
-## TLS ##
-
-tls_certificate_path: "/data/{{ SYNAPSE_SERVER_NAME }}.tls.crt"
-tls_private_key_path: "/data/{{ SYNAPSE_SERVER_NAME }}.tls.key"
-tls_dh_params_path: "/data/{{ SYNAPSE_SERVER_NAME }}.tls.dh"
-no_tls: {{ "True" if SYNAPSE_NO_TLS else "False" }}
-tls_fingerprints: []
-
-## Server ##
-
-server_name: "{{ SYNAPSE_SERVER_NAME }}"
-pid_file: /homeserver.pid
-web_client: False
-soft_file_limit: 0
-
-## Ports ##
-
-listeners:
-  {% if not SYNAPSE_NO_TLS %}
-  -
-    port: 8448
-    bind_addresses: ['0.0.0.0']
-    type: http
-    tls: true
-    x_forwarded: false
-    resources:
-      - names: [client]
-        compress: true
-      - names: [federation]  # Federation APIs
-        compress: false
-  {% endif %}
-
-  - port: 8008
-    tls: false
-    bind_addresses: ['0.0.0.0']
-    type: http
-    x_forwarded: false
-
-    resources:
-      - names: [client]
-        compress: true
-      - names: [federation]
-        compress: false
-
-## Database ##
-
-{% if POSTGRES_PASSWORD %}
-database:
-  name: "psycopg2"
-  args:
-    user: "{{ POSTGRES_USER or "synapse" }}"
-    password: "{{ POSTGRES_PASSWORD }}"
-    database: "{{ POSTGRES_DB or "synapse" }}"
-    host: "{{ POSTGRES_HOST or "db" }}"
-    port: "{{ POSTGRES_PORT or "5432" }}"
-    cp_min: 5
-    cp_max: 10
-{% else %}
-database:
-  name: "sqlite3"
-  args:
-    database: "/data/homeserver.db"
-{% endif %}
-
-## Performance ##
-
-event_cache_size: "{{ SYNAPSE_EVENT_CACHE_SIZE or "10K" }}"
-verbose: 0
-log_file: "/data/homeserver.log"
-log_config: "/compiled/log.config"
-
-## Ratelimiting ##
-
-rc_messages_per_second: 0.2
-rc_message_burst_count: 10.0
-federation_rc_window_size: 1000
-federation_rc_sleep_limit: 10
-federation_rc_sleep_delay: 500
-federation_rc_reject_limit: 50
-federation_rc_concurrent: 3
-
-## Files ##
-
-media_store_path: "/data/media"
-uploads_path: "/data/uploads"
-max_upload_size: "10M"
-max_image_pixels: "32M"
-dynamic_thumbnails: false
-
-# List of thumbnail to precalculate when an image is uploaded.
-thumbnail_sizes:
-- width: 32
-  height: 32
-  method: crop
-- width: 96
-  height: 96
-  method: crop
-- width: 320
-  height: 240
-  method: scale
-- width: 640
-  height: 480
-  method: scale
-- width: 800
-  height: 600
-  method: scale
-
-url_preview_enabled: False
-max_spider_size: "10M"
-
-## Captcha ##
-
-{% if SYNAPSE_RECAPTCHA_PUBLIC_KEY %}
-recaptcha_public_key: "{{ SYNAPSE_RECAPTCHA_PUBLIC_KEY }}"
-recaptcha_private_key: "{{ SYNAPSE_RECAPTCHA_PRIVATE_KEY }}"
-enable_registration_captcha: True
-recaptcha_siteverify_api: "https://www.google.com/recaptcha/api/siteverify"
-{% else %}
-recaptcha_public_key: "YOUR_PUBLIC_KEY"
-recaptcha_private_key: "YOUR_PRIVATE_KEY"
-enable_registration_captcha: False
-recaptcha_siteverify_api: "https://www.google.com/recaptcha/api/siteverify"
-{% endif %}
-
-## Turn ##
-
-{% if SYNAPSE_TURN_URIS %}
-turn_uris:
-{% for uri in SYNAPSE_TURN_URIS.split(',') %}    - "{{ uri }}"
-{% endfor %}
-turn_shared_secret: "{{ SYNAPSE_TURN_SECRET }}"
-turn_user_lifetime: "1h"
-turn_allow_guests: True
-{% else %}
-turn_uris: []
-turn_shared_secret: "YOUR_SHARED_SECRET"
-turn_user_lifetime: "1h"
-turn_allow_guests: True
-{% endif %}
-
-## Registration ##
-
-enable_registration: {{ "True" if SYNAPSE_ENABLE_REGISTRATION else "False" }}
-registration_shared_secret: "{{ SYNAPSE_REGISTRATION_SHARED_SECRET }}"
-bcrypt_rounds: 12
-allow_guest_access: {{ "True" if SYNAPSE_ALLOW_GUEST else "False" }}
-enable_group_creation: true
-
-# The list of identity servers trusted to verify third party
-# identifiers by this server.
-trusted_third_party_id_servers:
-    - matrix.org
-    - vector.im
-    - riot.im
-
-## Metrics ###
-
-{% if SYNAPSE_REPORT_STATS.lower() == "yes" %}
-enable_metrics: True
-report_stats: True
-{% else %}
-enable_metrics: False
-report_stats: False
-{% endif %}
-
-## API Configuration ##
-
-room_invite_state_types:
-    - "m.room.join_rules"
-    - "m.room.canonical_alias"
-    - "m.room.avatar"
-    - "m.room.name"
-
-{% if SYNAPSE_APPSERVICES %}
-app_service_config_files:
-{% for appservice in SYNAPSE_APPSERVICES %}    - "{{ appservice }}"
-{% endfor %}
-{% else %}
-app_service_config_files: []
-{% endif %}
-
-macaroon_secret_key: "{{ SYNAPSE_MACAROON_SECRET_KEY }}"
-expire_access_token: False
-
-## Signing Keys ##
-
-signing_key_path: "/data/{{ SYNAPSE_SERVER_NAME }}.signing.key"
-old_signing_keys: {}
-key_refresh_interval: "1d" # 1 Day.
-
-# The trusted servers to download signing keys from.
-perspectives:
-  servers:
-    "matrix.org":
-      verify_keys:
-        "ed25519:auto":
-          key: "Noi6WqcDj0QmPxCNQqgezwTlBKrfqehY1u2FyWP9uYw"
-
-password_config:
-   enabled: true
-
-{% if SYNAPSE_SMTP_HOST %}
-email:
-   enable_notifs: false
-   smtp_host: "{{ SYNAPSE_SMTP_HOST }}"
-   smtp_port: {{ SYNAPSE_SMTP_PORT or "25" }}
-   smtp_user: "{{ SYNAPSE_SMTP_USER }}"
-   smtp_pass: "{{ SYNAPSE_SMTP_PASSWORD }}"
-   require_transport_security: False
-   notif_from: "{{ SYNAPSE_SMTP_FROM or "hostmaster@" + SYNAPSE_SERVER_NAME }}"
-   app_name: Matrix
-   template_dir: res/templates
-   notif_template_html: notif_mail.html
-   notif_template_text: notif_mail.txt
-   notif_for_new_users: True
-   riot_base_url: "https://{{ SYNAPSE_SERVER_NAME }}"
-{% endif %}

contrib/docker/conf/log.config

@@ -1,29 +0,0 @@
-version: 1
-
-formatters:
-  precise:
-   format: '%(asctime)s - %(name)s - %(lineno)d - %(levelname)s - %(request)s- %(message)s'
-
-filters:
-  context:
-    (): synapse.util.logcontext.LoggingContextFilter
-    request: ""
-
-handlers:
-  console:
-    class: logging.StreamHandler
-    formatter: precise
-    filters: [context]
-
-loggers:
-    synapse:
-        level: {{ SYNAPSE_LOG_LEVEL or "WARNING" }}
-
-    synapse.storage.SQL:
-        # beware: increasing this to DEBUG will make synapse log sensitive
-        # information such as access tokens.
-        level: {{ SYNAPSE_LOG_LEVEL or "WARNING" }}
-
-root:
-    level: {{ SYNAPSE_LOG_LEVEL or "WARNING" }}
-    handlers: [console]

contrib/docker/docker-compose.yml

@@ -1,49 +0,0 @@
-# This compose file is compatible with Compose itself, it might need some
-# adjustments to run properly with stack.
-
-version: '3'
-
-services:
-
-  synapse:
-    image: docker.io/matrixdotorg/synapse:latest
-    # Since snyapse does not retry to connect to the database, restart upon
-    # failure
-    restart: unless-stopped
-    # See the readme for a full documentation of the environment settings
-    environment:
-      - SYNAPSE_SERVER_NAME=my.matrix.host
-      - SYNAPSE_REPORT_STATS=no
-      - SYNAPSE_ENABLE_REGISTRATION=yes
-      - SYNAPSE_LOG_LEVEL=INFO
-      - POSTGRES_PASSWORD=changeme
-    volumes:
-      # You may either store all the files in a local folder
-      - ./files:/data
-      # .. or you may split this between different storage points
-      # - ./files:/data
-      # - /path/to/ssd:/data/uploads
-      # - /path/to/large_hdd:/data/media
-    depends_on:
-      - db
-    # In order to expose Synapse, remove one of the following, you might for
-    # instance expose the TLS port directly:
-    ports:
-      - 8448:8448/tcp
-    # ... or use a reverse proxy, here is an example for traefik:
-    labels:
-      - traefik.enable=true
-      - traefik.frontend.rule=Host:my.matrix.Host
-      - traefik.port=8448
-
-  db:
-    image: docker.io/postgres:10-alpine
-    # Change that password, of course!
-    environment:
-      - POSTGRES_USER=synapse
-      - POSTGRES_PASSWORD=changeme
-    volumes:
-      # You may store the database tables in a local folder..
-      - ./schemas:/var/lib/postgresql/data
-      # .. or store them on some high performance storage for better results
-      # - /path/to/ssd/storage:/var/lib/postfesql/data

contrib/docker/start.py

@@ -1,66 +0,0 @@
-#!/usr/local/bin/python
-
-import jinja2
-import os
-import sys
-import subprocess
-import glob
-
-# Utility functions
-convert = lambda src, dst, environ: open(dst, "w").write(jinja2.Template(open(src).read()).render(**environ))
-
-def check_arguments(environ, args):
-    for argument in args:
-        if argument not in environ:
-            print("Environment variable %s is mandatory, exiting." % argument)
-            sys.exit(2)
-
-def generate_secrets(environ, secrets):
-    for name, secret in secrets.items():
-        if secret not in environ:
-            filename = "/data/%s.%s.key" % (environ["SYNAPSE_SERVER_NAME"], name)
-            if os.path.exists(filename):
-                with open(filename) as handle: value = handle.read()
-            else:
-                print("Generating a random secret for {}".format(name))
-                value = os.urandom(32).encode("hex")
-                with open(filename, "w") as handle: handle.write(value)
-            environ[secret] = value
-
-# Prepare the configuration
-mode = sys.argv[1] if len(sys.argv) > 1 else None
-environ = os.environ.copy()
-ownership = "{}:{}".format(environ.get("UID", 991), environ.get("GID", 991))
-args = ["python", "-m", "synapse.app.homeserver"]
-
-# In generate mode, generate a configuration, missing keys, then exit
-if mode == "generate":
-    check_arguments(environ, ("SYNAPSE_SERVER_NAME", "SYNAPSE_REPORT_STATS", "SYNAPSE_CONFIG_PATH"))
-    args += [
-        "--server-name", environ["SYNAPSE_SERVER_NAME"],
-        "--report-stats", environ["SYNAPSE_REPORT_STATS"],
-        "--config-path", environ["SYNAPSE_CONFIG_PATH"],
-        "--generate-config"
-    ]
-    os.execv("/usr/local/bin/python", args)
-
-# In normal mode, generate missing keys if any, then run synapse
-else:
-    # Parse the configuration file
-    if "SYNAPSE_CONFIG_PATH" in environ:
-        args += ["--config-path", environ["SYNAPSE_CONFIG_PATH"]]
-    else:
-        check_arguments(environ, ("SYNAPSE_SERVER_NAME", "SYNAPSE_REPORT_STATS"))
-        generate_secrets(environ, {
-            "registration": "SYNAPSE_REGISTRATION_SHARED_SECRET",
-            "macaroon": "SYNAPSE_MACAROON_SECRET_KEY"
-        })
-        environ["SYNAPSE_APPSERVICES"] = glob.glob("/data/appservices/*.yaml")
-        if not os.path.exists("/compiled"): os.mkdir("/compiled")
-        convert("/conf/homeserver.yaml", "/compiled/homeserver.yaml", environ)
-        convert("/conf/log.config", "/compiled/log.config", environ)
-        subprocess.check_output(["chown", "-R", ownership, "/data"])
-        args += ["--config-path", "/compiled/homeserver.yaml"]
-    # Generate missing keys and start synapse
-    subprocess.check_output(args + ["--generate-keys"])
-    os.execv("/sbin/su-exec", ["su-exec", ownership] + args)

docs/postgres.rst

@@ -6,13 +6,7 @@ Postgres version 9.4 or later is known to work.
 Set up database
 ===============
 
-Assuming your PostgreSQL database user is called ``postgres``, create a user
-``synapse_user`` with::
-
- su - postgres
- createuser --pwprompt synapse_user
-
-The PostgreSQL database used *must* have the correct encoding set, otherwise it
+The PostgreSQL database used *must* have the correct encoding set, otherwise
 would not be able to store UTF8 strings. To create a database with the correct
 encoding use, e.g.::
 
@@ -52,8 +46,8 @@ As with Debian/Ubuntu, postgres support depends on the postgres python connector
 Synapse config
 ==============
 
-When you are ready to start using PostgreSQL, edit the ``database`` section in
-your config file to match the following lines::
+When you are ready to start using PostgreSQL, add the following line to your
+config file::
 
     database:
         name: psycopg2
@@ -102,12 +96,9 @@ complete, restart synapse.  For instance::
     cp homeserver.db homeserver.db.snapshot
     ./synctl start
 
-Copy the old config file into a new config file::
-
-    cp homeserver.yaml homeserver-postgres.yaml
-
-Edit the database section as described in the section *Synapse config* above
-and with the SQLite snapshot located at ``homeserver.db.snapshot`` simply run::
+Assuming your new config file (as described in the section *Synapse config*)
+is named ``homeserver-postgres.yaml`` and the SQLite snapshot is at
+``homeserver.db.snapshot`` then simply run::
 
     synapse_port_db --sqlite-database homeserver.db.snapshot \
         --postgres-config homeserver-postgres.yaml
@@ -126,11 +117,6 @@ run::
         --postgres-config homeserver-postgres.yaml
 
 Once that has completed, change the synapse config to point at the PostgreSQL
-database configuration file ``homeserver-postgres.yaml``:
-
-    ./synctl stop
-    mv homeserver.yaml homeserver-old-sqlite.yaml 
-    mv homeserver-postgres.yaml homeserver.yaml 
-    ./synctl start
-
-Synapse should now be running against PostgreSQL.
+database configuration file ``homeserver-postgres.yaml`` (i.e. rename it to 
+``homeserver.yaml``) and restart synapse. Synapse should now be running against
+PostgreSQL.

docs/privacy_policy_templates/README.md

@@ -1,23 +0,0 @@
-If enabling the 'consent' resource in synapse, you will need some templates
-for the HTML to be served to the user. This directory contains very simple
-examples of the sort of thing that can be done.
-
-You'll need to add this sort of thing to your homeserver.yaml:
-
-```
-form_secret: <unique but arbitrary secret>
-
-user_consent:
-  template_dir: docs/privacy_policy_templates
-  default_version: 1.0
-```
-
-You should then be able to enable the `consent` resource under a `listener`
-entry. For example:
-
-```
-listeners:
-  - port: 8008
-    resources:
-      - names: [client, consent]
-```

docs/privacy_policy_templates/en/1.0.html

@@ -1,17 +0,0 @@
-<!doctype html>
-<html lang="en">
-  <head>
-    <title>Matrix.org Privacy policy</title>
-  </head>
-  <body>
-    <p>
-      All your base are belong to us.
-    </p>
-    <form method="post" action="consent">
-      <input type="hidden" name="v" value="{{version}}"/>
-      <input type="hidden" name="u" value="{{user}}"/>
-      <input type="hidden" name="h" value="{{userhmac}}"/>
-      <input type="submit" value="Sure thing!"/>
-    </form>
-  </body>
-</html>

docs/privacy_policy_templates/en/success.html

@@ -1,11 +0,0 @@
-<!doctype html>
-<html lang="en">
-  <head>
-    <title>Matrix.org Privacy policy</title>
-  </head>
-  <body>
-    <p>
-      Sweet.
-    </p>
-  </body>
-</html>

scripts-dev/nuke-room-from-db.sh

@@ -6,19 +6,9 @@
 
 ## Do not run it lightly.
 
-set -e
-
-if [ "$1" == "-h" ] || [ "$1" == "" ]; then
-  echo "Call with ROOM_ID as first option and then pipe it into the database. So for instance you might run"
-  echo " nuke-room-from-db.sh <room_id> | sqlite3 homeserver.db"
-  echo "or"
-  echo " nuke-room-from-db.sh <room_id> | psql --dbname=synapse"
-  exit
-fi
-
 ROOMID="$1"
 
-cat <<EOF
+sqlite3 homeserver.db <<EOF
 DELETE FROM event_forward_extremities WHERE room_id = '$ROOMID';
 DELETE FROM event_backward_extremities WHERE room_id = '$ROOMID';
 DELETE FROM event_edges WHERE room_id = '$ROOMID';
@@ -39,7 +29,7 @@ DELETE FROM state_groups WHERE room_id = '$ROOMID';
 DELETE FROM state_groups_state WHERE room_id = '$ROOMID';
 DELETE FROM receipts_graph WHERE room_id = '$ROOMID';
 DELETE FROM receipts_linearized WHERE room_id = '$ROOMID';
-DELETE FROM event_search WHERE room_id = '$ROOMID'; 
+DELETE FROM event_search_content WHERE c1room_id = '$ROOMID';
 DELETE FROM guest_access WHERE room_id = '$ROOMID';
 DELETE FROM history_visibility WHERE room_id = '$ROOMID';
 DELETE FROM room_tags WHERE room_id = '$ROOMID';

synapse/__init__.py

@@ -16,4 +16,4 @@
 """ This is a reference implementation of a Matrix home server.
 """
 
-__version__ = "0.29.1"
+__version__ = "0.28.1"

synapse/app/appservice.py

@@ -74,7 +74,6 @@ class AppserviceServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/client_reader.py

@@ -98,7 +98,6 @@ class ClientReaderServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/event_creator.py

@@ -114,7 +114,6 @@ class EventCreatorServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/federation_reader.py

@@ -87,7 +87,6 @@ class FederationReaderServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/federation_sender.py

@@ -101,7 +101,6 @@ class FederationSenderServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/frontend_proxy.py

@@ -152,7 +152,6 @@ class FrontendProxyServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/homeserver.py

@@ -41,7 +41,6 @@ from synapse.python_dependencies import CONDITIONAL_REQUIREMENTS, \
 from synapse.replication.http import ReplicationRestResource, REPLICATION_PREFIX
 from synapse.replication.tcp.resource import ReplicationStreamProtocolFactory
 from synapse.rest import ClientRestResource
-from synapse.rest.consent.consent_resource import ConsentResource
 from synapse.rest.key.v1.server_key_resource import LocalKey
 from synapse.rest.key.v2 import KeyApiV2Resource
 from synapse.rest.media.v0.content_repository import ContentRepoResource
@@ -141,7 +140,6 @@ class SynapseHomeServer(HomeServer):
                     site_tag,
                     listener_config,
                     root_resource,
-                    self.version_string,
                 ),
                 self.tls_server_context_factory,
             )
@@ -155,7 +153,6 @@ class SynapseHomeServer(HomeServer):
                     site_tag,
                     listener_config,
                     root_resource,
-                    self.version_string,
                 )
             )
         logger.info("Synapse now listening on port %d", port)
@@ -185,14 +182,6 @@ class SynapseHomeServer(HomeServer):
                 "/_matrix/client/versions": client_resource,
             })
 
-        if name == "consent":
-            consent_resource = ConsentResource(self)
-            if compress:
-                consent_resource = gz_wrap(consent_resource)
-            resources.update({
-                "/_matrix/consent": consent_resource,
-            })
-
         if name == "federation":
             resources.update({
                 FEDERATION_PREFIX: TransportLayerServer(self),
@@ -484,14 +473,6 @@ def run(hs):
                 " changes across releases."
             )
 
-    def generate_user_daily_visit_stats():
-        hs.get_datastore().generate_user_daily_visits()
-
-    # Rather than update on per session basis, batch up the requests.
-    # If you increase the loop period, the accuracy of user_daily_visits
-    # table will decrease
-    clock.looping_call(generate_user_daily_visit_stats, 5 * 60 * 1000)
-
     if hs.config.report_stats:
         logger.info("Scheduling stats reporting for 3 hour intervals")
         clock.looping_call(phone_stats_home, 3 * 60 * 60 * 1000)

synapse/app/media_repository.py

@@ -94,7 +94,6 @@ class MediaRepositoryServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/pusher.py

@@ -104,7 +104,6 @@ class PusherServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/synchrotron.py

@@ -281,7 +281,6 @@ class SynchrotronServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/app/user_dir.py

@@ -126,7 +126,6 @@ class UserDirectoryServer(HomeServer):
                 site_tag,
                 listener_config,
                 root_resource,
-                self.version_string,
             )
         )
 

synapse/config/__init__.py

@@ -12,9 +12,3 @@
 # 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 ._base import ConfigError
-
-# export ConfigError if somebody does import *
-# this is largely a fudge to stop PEP8 moaning about the import
-__all__ = ["ConfigError"]

synapse/config/consent_config.py

@@ -1,42 +0,0 @@
-# -*- 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 ._base import Config
-
-DEFAULT_CONFIG = """\
-# User Consent configuration
-#
-# uncomment and configure if enabling the 'consent' resource under 'listeners'.
-#
-# 'template_dir' gives the location of the templates for the HTML forms.
-# This directory should contain one subdirectory per language (eg, 'en', 'fr'),
-# and each language directory should contain the policy document (named as
-# '<version>.html') and a success page (success.html).
-#
-# 'default_version' gives the version of the policy document to serve up if
-# there is no 'v' parameter.
-#
-# user_consent:
-#   template_dir: res/templates/privacy
-#   default_version: 1.0
-"""
-
-
-class ConsentConfig(Config):
-    def read_config(self, config):
-        self.consent_config = config.get("user_consent")
-
-    def default_config(self, **kwargs):
-        return DEFAULT_CONFIG

synapse/config/homeserver.py

@@ -1,6 +1,5 @@
 # -*- coding: utf-8 -*-
 # Copyright 2014-2016 OpenMarket Ltd
-# 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.
@@ -13,6 +12,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 .tls import TlsConfig
 from .server import ServerConfig
 from .logger import LoggingConfig
@@ -37,8 +37,6 @@ from .push import PushConfig
 from .spam_checker import SpamCheckerConfig
 from .groups import GroupsConfig
 from .user_directory import UserDirectoryConfig
-from .consent_config import ConsentConfig
-from .server_notices_config import ServerNoticesConfig
 
 
 class HomeServerConfig(TlsConfig, ServerConfig, DatabaseConfig, LoggingConfig,
@@ -47,15 +45,12 @@ class HomeServerConfig(TlsConfig, ServerConfig, DatabaseConfig, LoggingConfig,
                        AppServiceConfig, KeyConfig, SAML2Config, CasConfig,
                        JWTConfig, PasswordConfig, EmailConfig,
                        WorkerConfig, PasswordAuthProviderConfig, PushConfig,
-                       SpamCheckerConfig, GroupsConfig, UserDirectoryConfig,
-                       ConsentConfig,
-                       ServerNoticesConfig,
-                       ):
+                       SpamCheckerConfig, GroupsConfig, UserDirectoryConfig,):
     pass
 
 
 if __name__ == '__main__':
     import sys
     sys.stdout.write(
-        HomeServerConfig().generate_config(sys.argv[1], sys.argv[2], True)[0]
+        HomeServerConfig().generate_config(sys.argv[1], sys.argv[2])[0]
     )

synapse/config/key.py

@@ -59,20 +59,14 @@ class KeyConfig(Config):
 
         self.expire_access_token = config.get("expire_access_token", False)
 
-        # a secret which is used to calculate HMACs for form values, to stop
-        # falsification of values
-        self.form_secret = config.get("form_secret", None)
-
     def default_config(self, config_dir_path, server_name, is_generating_file=False,
                        **kwargs):
         base_key_name = os.path.join(config_dir_path, server_name)
 
         if is_generating_file:
             macaroon_secret_key = random_string_with_symbols(50)
-            form_secret = '"%s"' % random_string_with_symbols(50)
         else:
             macaroon_secret_key = None
-            form_secret = 'null'
 
         return """\
         macaroon_secret_key: "%(macaroon_secret_key)s"
@@ -80,10 +74,6 @@ class KeyConfig(Config):
         # Used to enable access token expiration.
         expire_access_token: False
 
-        # a secret which is used to calculate HMACs for form values, to stop
-        # falsification of values
-        form_secret: %(form_secret)s
-
         ## Signing Keys ##
 
         # Path to the signing key to sign messages with

synapse/config/server_notices_config.py

@@ -1,77 +0,0 @@
-# -*- 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 ._base import Config
-from synapse.types import UserID
-
-DEFAULT_CONFIG = """\
-# Server Notices room configuration
-#
-# Uncomment this section to enable a room which can be used to send notices
-# from the server to users. It is a special room which cannot be left; notices
-# come from a special "notices" user id.
-#
-# If you uncomment this section, you *must* define the system_mxid_localpart
-# setting, which defines the id of the user which will be used to send the
-# notices.
-#
-# It's also possible to override the room name, or the display name of the
-# "notices" user.
-#
-# server_notices:
-#   system_mxid_localpart: notices
-#   system_mxid_display_name: "Server Notices"
-#   room_name: "Server Notices"
-"""
-
-
-class ServerNoticesConfig(Config):
-    """Configuration for the server notices room.
-
-    Attributes:
-        server_notices_mxid (str|None):
-            The MXID to use for server notices.
-            None if server notices are not enabled.
-
-        server_notices_mxid_display_name (str|None):
-            The display name to use for the server notices user.
-            None if server notices are not enabled.
-
-        server_notices_room_name (str|None):
-            The name to use for the server notices room.
-            None if server notices are not enabled.
-    """
-    def __init__(self):
-        super(ServerNoticesConfig, self).__init__()
-        self.server_notices_mxid = None
-        self.server_notices_mxid_display_name = None
-        self.server_notices_room_name = None
-
-    def read_config(self, config):
-        c = config.get("server_notices")
-        if c is None:
-            return
-
-        mxid_localpart = c['system_mxid_localpart']
-        self.server_notices_mxid = UserID(
-            mxid_localpart, self.server_name,
-        ).to_string()
-        self.server_notices_mxid_display_name = c.get(
-            'system_mxid_display_name', 'Server Notices',
-        )
-        # todo: i18n
-        self.server_notices_room_name = c.get('room_name', "Server Notices")
-
-    def default_config(self, **kwargs):
-        return DEFAULT_CONFIG

synapse/federation/units.py

@@ -74,6 +74,8 @@ class Transaction(JsonEncodedObject):
         "previous_ids",
         "pdus",
         "edus",
+        "transaction_id",
+        "destination",
         "pdu_failures",
     ]
 

synapse/handlers/__init__.py

@@ -14,7 +14,9 @@
 # limitations under the License.
 
 from .register import RegistrationHandler
-from .room import RoomContextHandler
+from .room import (
+    RoomCreationHandler, RoomContextHandler,
+)
 from .message import MessageHandler
 from .federation import FederationHandler
 from .directory import DirectoryHandler
@@ -45,6 +47,7 @@ class Handlers(object):
     def __init__(self, hs):
         self.registration_handler = RegistrationHandler(hs)
         self.message_handler = MessageHandler(hs)
+        self.room_creation_handler = RoomCreationHandler(hs)
         self.federation_handler = FederationHandler(hs)
         self.directory_handler = DirectoryHandler(hs)
         self.admin_handler = AdminHandler(hs)

synapse/handlers/deactivate_account.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-# Copyright 2017, 2018 New Vector Ltd
+# Copyright 2017 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.
@@ -12,11 +12,9 @@
 # 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 twisted.internet import defer, reactor
+from twisted.internet import defer
 
 from ._base import BaseHandler
-from synapse.types import UserID, create_requester
-from synapse.util.logcontext import run_in_background
 
 import logging
 
@@ -29,14 +27,6 @@ class DeactivateAccountHandler(BaseHandler):
         super(DeactivateAccountHandler, self).__init__(hs)
         self._auth_handler = hs.get_auth_handler()
         self._device_handler = hs.get_device_handler()
-        self._room_member_handler = hs.get_room_member_handler()
-
-        # Flag that indicates whether the process to part users from rooms is running
-        self._user_parter_running = False
-
-        # Start the user parter loop so it can resume parting users from rooms where
-        # it left off (if it has work left to do).
-        reactor.callWhenRunning(self._start_user_parting)
 
     @defer.inlineCallbacks
     def deactivate_account(self, user_id):
@@ -60,70 +50,3 @@ class DeactivateAccountHandler(BaseHandler):
 
         yield self.store.user_delete_threepids(user_id)
         yield self.store.user_set_password_hash(user_id, None)
-
-        # Add the user to a table of users penpding deactivation (ie.
-        # removal from all the rooms they're a member of)
-        yield self.store.add_user_pending_deactivation(user_id)
-
-        # Now start the process that goes through that list and
-        # parts users from rooms (if it isn't already running)
-        self._start_user_parting()
-
-    def _start_user_parting(self):
-        """
-        Start the process that goes through the table of users
-        pending deactivation, if it isn't already running.
-
-        Returns:
-            None
-        """
-        if not self._user_parter_running:
-            run_in_background(self._user_parter_loop)
-
-    @defer.inlineCallbacks
-    def _user_parter_loop(self):
-        """Loop that parts deactivated users from rooms
-
-        Returns:
-            None
-        """
-        self._user_parter_running = True
-        logger.info("Starting user parter")
-        try:
-            while True:
-                user_id = yield self.store.get_user_pending_deactivation()
-                if user_id is None:
-                    break
-                logger.info("User parter parting %r", user_id)
-                yield self._part_user(user_id)
-                yield self.store.del_user_pending_deactivation(user_id)
-                logger.info("User parter finished parting %r", user_id)
-            logger.info("User parter finished: stopping")
-        finally:
-            self._user_parter_running = False
-
-    @defer.inlineCallbacks
-    def _part_user(self, user_id):
-        """Causes the given user_id to leave all the rooms they're joined to
-
-        Returns:
-            None
-        """
-        user = UserID.from_string(user_id)
-
-        rooms_for_user = yield self.store.get_rooms_for_user(user_id)
-        for room_id in rooms_for_user:
-            logger.info("User parter parting %r from %r", user_id, room_id)
-            try:
-                yield self._room_member_handler.update_membership(
-                    create_requester(user),
-                    user,
-                    room_id,
-                    "leave",
-                    ratelimit=False,
-                )
-            except Exception:
-                logger.exception(
-                    "Failed to part user %r from room %r: ignoring and continuing",
-                    user_id, room_id,
-                )

synapse/handlers/federation.py

@@ -81,7 +81,6 @@ class FederationHandler(BaseHandler):
         self.pusher_pool = hs.get_pusherpool()
         self.spam_checker = hs.get_spam_checker()
         self.event_creation_handler = hs.get_event_creation_handler()
-        self._server_notices_mxid = hs.config.server_notices_mxid
 
         # When joining a room we need to queue any events for that room up
         self.room_queues = {}
@@ -1181,13 +1180,6 @@ class FederationHandler(BaseHandler):
         if not self.is_mine_id(event.state_key):
             raise SynapseError(400, "The invite event must be for this server")
 
-        # block any attempts to invite the server notices mxid
-        if event.state_key == self._server_notices_mxid:
-            raise SynapseError(
-                http_client.FORBIDDEN,
-                "Cannot invite this user",
-            )
-
         event.internal_metadata.outlier = True
         event.internal_metadata.invite_from_remote = True
 

synapse/handlers/initial_sync.py

@@ -181,8 +181,8 @@ class InitialSyncHandler(BaseHandler):
                     self.store, user_id, messages
                 )
 
-                start_token = now_token.copy_and_replace("room_key", token)
-                end_token = now_token.copy_and_replace("room_key", room_end_token)
+                start_token = now_token.copy_and_replace("room_key", token[0])
+                end_token = now_token.copy_and_replace("room_key", token[1])
                 time_now = self.clock.time_msec()
 
                 d["messages"] = {
@@ -325,8 +325,8 @@ class InitialSyncHandler(BaseHandler):
             self.store, user_id, messages, is_peeking=is_peeking
         )
 
-        start_token = StreamToken.START.copy_and_replace("room_key", token)
-        end_token = StreamToken.START.copy_and_replace("room_key", stream_token)
+        start_token = StreamToken.START.copy_and_replace("room_key", token[0])
+        end_token = StreamToken.START.copy_and_replace("room_key", token[1])
 
         time_now = self.clock.time_msec()
 
@@ -408,8 +408,8 @@ class InitialSyncHandler(BaseHandler):
             self.store, user_id, messages, is_peeking=is_peeking,
         )
 
-        start_token = now_token.copy_and_replace("room_key", token)
-        end_token = now_token
+        start_token = now_token.copy_and_replace("room_key", token[0])
+        end_token = now_token.copy_and_replace("room_key", token[1])
 
         time_now = self.clock.time_msec()
 

synapse/handlers/message.py

@@ -86,14 +86,14 @@ class MessageHandler(BaseHandler):
         # map from purge id to PurgeStatus
         self._purges_by_id = {}
 
-    def start_purge_history(self, room_id, token,
+    def start_purge_history(self, room_id, topological_ordering,
                             delete_local_events=False):
         """Start off a history purge on a room.
 
         Args:
             room_id (str): The room to purge from
 
-            token (str): topological token to delete events before
+            topological_ordering (int): minimum topo ordering to preserve
             delete_local_events (bool): True to delete local events as well as
                 remote ones
 
@@ -115,19 +115,19 @@ class MessageHandler(BaseHandler):
         self._purges_by_id[purge_id] = PurgeStatus()
         run_in_background(
             self._purge_history,
-            purge_id, room_id, token, delete_local_events,
+            purge_id, room_id, topological_ordering, delete_local_events,
         )
         return purge_id
 
     @defer.inlineCallbacks
-    def _purge_history(self, purge_id, room_id, token,
+    def _purge_history(self, purge_id, room_id, topological_ordering,
                        delete_local_events):
         """Carry out a history purge on a room.
 
         Args:
             purge_id (str): The id for this purge
             room_id (str): The room to purge from
-            token (str): topological token to delete events before
+            topological_ordering (int): minimum topo ordering to preserve
             delete_local_events (bool): True to delete local events as well as
                 remote ones
 
@@ -138,7 +138,7 @@ class MessageHandler(BaseHandler):
         try:
             with (yield self.pagination_lock.write(room_id)):
                 yield self.store.purge_history(
-                    room_id, token, delete_local_events,
+                    room_id, topological_ordering, delete_local_events,
                 )
             logger.info("[purge] complete")
             self._purges_by_id[purge_id].status = PurgeStatus.STATUS_COMPLETE

synapse/handlers/presence.py

@@ -463,6 +463,61 @@ class PresenceHandler(object):
             syncing_user_ids.update(user_ids)
         return syncing_user_ids
 
+    @defer.inlineCallbacks
+    def update_external_syncs(self, process_id, syncing_user_ids):
+        """Update the syncing users for an external process
+
+        Args:
+            process_id(str): An identifier for the process the users are
+                syncing against. This allows synapse to process updates
+                as user start and stop syncing against a given process.
+            syncing_user_ids(set(str)): The set of user_ids that are
+                currently syncing on that server.
+        """
+
+        # Grab the previous list of user_ids that were syncing on that process
+        prev_syncing_user_ids = (
+            self.external_process_to_current_syncs.get(process_id, set())
+        )
+        # Grab the current presence state for both the users that are syncing
+        # now and the users that were syncing before this update.
+        prev_states = yield self.current_state_for_users(
+            syncing_user_ids | prev_syncing_user_ids
+        )
+        updates = []
+        time_now_ms = self.clock.time_msec()
+
+        # For each new user that is syncing check if we need to mark them as
+        # being online.
+        for new_user_id in syncing_user_ids - prev_syncing_user_ids:
+            prev_state = prev_states[new_user_id]
+            if prev_state.state == PresenceState.OFFLINE:
+                updates.append(prev_state.copy_and_replace(
+                    state=PresenceState.ONLINE,
+                    last_active_ts=time_now_ms,
+                    last_user_sync_ts=time_now_ms,
+                ))
+            else:
+                updates.append(prev_state.copy_and_replace(
+                    last_user_sync_ts=time_now_ms,
+                ))
+
+        # For each user that is still syncing or stopped syncing update the
+        # last sync time so that we will correctly apply the grace period when
+        # they stop syncing.
+        for old_user_id in prev_syncing_user_ids:
+            prev_state = prev_states[old_user_id]
+            updates.append(prev_state.copy_and_replace(
+                last_user_sync_ts=time_now_ms,
+            ))
+
+        yield self._update_states(updates)
+
+        # Update the last updated time for the process. We expire the entries
+        # if we don't receive an update in the given timeframe.
+        self.external_process_last_updated_ms[process_id] = self.clock.time_msec()
+        self.external_process_to_current_syncs[process_id] = syncing_user_ids
+
     @defer.inlineCallbacks
     def update_external_syncs_row(self, process_id, user_id, is_syncing, sync_time_msec):
         """Update the syncing users for an external process as a delta.

synapse/handlers/register.py

@@ -34,11 +34,6 @@ logger = logging.getLogger(__name__)
 class RegistrationHandler(BaseHandler):
 
     def __init__(self, hs):
-        """
-
-        Args:
-            hs (synapse.server.HomeServer):
-        """
         super(RegistrationHandler, self).__init__(hs)
 
         self.auth = hs.get_auth()
@@ -54,7 +49,6 @@ class RegistrationHandler(BaseHandler):
         self._generate_user_id_linearizer = Linearizer(
             name="_generate_user_id_linearizer",
         )
-        self._server_notices_mxid = hs.config.server_notices_mxid
 
     @defer.inlineCallbacks
     def check_username(self, localpart, guest_access_token=None,
@@ -344,14 +338,6 @@ class RegistrationHandler(BaseHandler):
             yield identity_handler.bind_threepid(c, user_id)
 
     def check_user_id_not_appservice_exclusive(self, user_id, allowed_appservice=None):
-        # don't allow people to register the server notices mxid
-        if self._server_notices_mxid is not None:
-            if user_id == self._server_notices_mxid:
-                raise SynapseError(
-                    400, "This user ID is reserved.",
-                    errcode=Codes.EXCLUSIVE
-                )
-
         # valid user IDs must not clash with any user ID namespaces claimed by
         # application services.
         services = self.store.get_app_services()

synapse/handlers/room.py

@@ -68,27 +68,14 @@ class RoomCreationHandler(BaseHandler):
         self.event_creation_handler = hs.get_event_creation_handler()
 
     @defer.inlineCallbacks
-    def create_room(self, requester, config, ratelimit=True,
-                    creator_join_profile=None):
+    def create_room(self, requester, config, ratelimit=True):
         """ Creates a new room.
 
         Args:
-            requester (synapse.types.Requester):
-                The user who requested the room creation.
+            requester (Requester): The user who requested the room creation.
             config (dict) : A dict of configuration options.
-            ratelimit (bool): set to False to disable the rate limiter
-
-            creator_join_profile (dict|None):
-                Set to override the displayname and avatar for the creating
-                user in this room. If unset, displayname and avatar will be
-                derived from the user's profile. If set, should contain the
-                values to go in the body of the 'join' event (typically
-                `avatar_url` and/or `displayname`.
-
         Returns:
-            Deferred[dict]:
-                a dict containing the keys `room_id` and, if an alias was
-                requested, `room_alias`.
+            The new room ID.
         Raises:
             SynapseError if the room ID couldn't be stored, or something went
             horribly wrong.
@@ -189,8 +176,7 @@ class RoomCreationHandler(BaseHandler):
             initial_state=initial_state,
             creation_content=creation_content,
             room_alias=room_alias,
-            power_level_content_override=config.get("power_level_content_override", {}),
-            creator_join_profile=creator_join_profile,
+            power_level_content_override=config.get("power_level_content_override", {})
         )
 
         if "name" in config:
@@ -270,7 +256,6 @@ class RoomCreationHandler(BaseHandler):
             creation_content,
             room_alias,
             power_level_content_override,
-            creator_join_profile,
     ):
         def create(etype, content, **kwargs):
             e = {
@@ -314,7 +299,6 @@ class RoomCreationHandler(BaseHandler):
             room_id,
             "join",
             ratelimit=False,
-            content=creator_join_profile,
         )
 
         # We treat the power levels override specially as this needs to be one

synapse/handlers/room_member.py

@@ -17,14 +17,11 @@
 import abc
 import logging
 
-from six.moves import http_client
-
 from signedjson.key import decode_verify_key_bytes
 from signedjson.sign import verify_signed_json
 from twisted.internet import defer
 from unpaddedbase64 import decode_base64
 
-import synapse.server
 import synapse.types
 from synapse.api.constants import (
     EventTypes, Membership,
@@ -49,11 +46,6 @@ class RoomMemberHandler(object):
     __metaclass__ = abc.ABCMeta
 
     def __init__(self, hs):
-        """
-
-        Args:
-            hs (synapse.server.HomeServer):
-        """
         self.hs = hs
         self.store = hs.get_datastore()
         self.auth = hs.get_auth()
@@ -71,7 +63,6 @@ class RoomMemberHandler(object):
 
         self.clock = hs.get_clock()
         self.spam_checker = hs.get_spam_checker()
-        self._server_notices_mxid = self.config.server_notices_mxid
 
     @abc.abstractmethod
     def _remote_join(self, requester, remote_room_hosts, room_id, user, content):
@@ -298,36 +289,12 @@ class RoomMemberHandler(object):
             is_blocked = yield self.store.is_room_blocked(room_id)
             if is_blocked:
                 raise SynapseError(403, "This room has been blocked on this server")
-        else:
-            # we don't allow people to reject invites to, or leave, the
-            # server notice room.
-            is_blocked = yield self._is_server_notice_room(room_id)
-            if is_blocked:
-                raise SynapseError(
-                    http_client.FORBIDDEN,
-                    "You cannot leave this room",
-                )
-
-        if effective_membership_state == Membership.INVITE:
-            # block any attempts to invite the server notices mxid
-            if target.to_string() == self._server_notices_mxid:
-                raise SynapseError(
-                    http_client.FORBIDDEN,
-                    "Cannot invite this user",
-                )
 
+        if effective_membership_state == "invite":
             block_invite = False
-
-            if (self._server_notices_mxid is not None and
-                    requester.user.to_string() == self._server_notices_mxid):
-                # allow the server notices mxid to send invites
-                is_requester_admin = True
-
-            else:
-                is_requester_admin = yield self.auth.is_server_admin(
-                    requester.user,
-                )
-
+            is_requester_admin = yield self.auth.is_server_admin(
+                requester.user,
+            )
             if not is_requester_admin:
                 if self.config.block_non_admin_invites:
                     logger.info(
@@ -877,13 +844,6 @@ class RoomMemberHandler(object):
 
         defer.returnValue(False)
 
-    @defer.inlineCallbacks
-    def _is_server_notice_room(self, room_id):
-        if self._server_notices_mxid is None:
-            defer.returnValue(False)
-        user_ids = yield self.store.get_users_in_room(room_id)
-        defer.returnValue(self._server_notices_mxid in user_ids)
-
 
 class RoomMemberMasterHandler(RoomMemberHandler):
     def __init__(self, hs):

synapse/handlers/sync.py

@@ -354,24 +354,12 @@ class SyncHandler(object):
                 since_key = since_token.room_key
 
             while limited and len(recents) < timeline_limit and max_repeat:
-                # If we have a since_key then we are trying to get any events
-                # that have happened since `since_key` up to `end_key`, so we
-                # can just use `get_room_events_stream_for_room`.
-                # Otherwise, we want to return the last N events in the room
-                # in toplogical ordering.
-                if since_key:
-                    events, end_key = yield self.store.get_room_events_stream_for_room(
-                        room_id,
-                        limit=load_limit + 1,
-                        from_key=since_key,
-                        to_key=end_key,
-                    )
-                else:
-                    events, end_key = yield self.store.get_recent_events_for_room(
-                        room_id,
-                        limit=load_limit + 1,
-                        end_token=end_key,
-                    )
+                events, end_key = yield self.store.get_room_events_stream_for_room(
+                    room_id,
+                    limit=load_limit + 1,
+                    from_key=since_key,
+                    to_key=end_key,
+                )
                 loaded_recents = sync_config.filter_collection.filter_room_timeline(
                     events
                 )
@@ -441,7 +429,7 @@ class SyncHandler(object):
         Returns:
             A Deferred map from ((type, state_key)->Event)
         """
-        last_events, _ = yield self.store.get_recent_events_for_room(
+        last_events, token = yield self.store.get_recent_events_for_room(
             room_id, end_token=stream_position.room_key, limit=1,
         )
 

synapse/http/additional_resource.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from synapse.http.server import wrap_json_request_handler
+from synapse.http.server import wrap_request_handler
 from twisted.web.resource import Resource
 from twisted.web.server import NOT_DONE_YET
 
@@ -42,13 +42,14 @@ class AdditionalResource(Resource):
         Resource.__init__(self)
         self._handler = handler
 
-        # required by the request_handler wrapper
+        # these are required by the request_handler wrapper
+        self.version_string = hs.version_string
         self.clock = hs.get_clock()
 
     def render(self, request):
         self._async_render(request)
         return NOT_DONE_YET
 
-    @wrap_json_request_handler
+    @wrap_request_handler
     def _async_render(self, request):
         return self._handler(request)

synapse/http/request_metrics.py

@@ -1,149 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2014-2016 OpenMarket Ltd
-# 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
-
-import synapse.metrics
-from synapse.util.logcontext import LoggingContext
-
-logger = logging.getLogger(__name__)
-
-metrics = synapse.metrics.get_metrics_for("synapse.http.server")
-
-# total number of responses served, split by method/servlet/tag
-response_count = metrics.register_counter(
-    "response_count",
-    labels=["method", "servlet", "tag"],
-    alternative_names=(
-        # the following are all deprecated aliases for the same metric
-        metrics.name_prefix + x for x in (
-            "_requests",
-            "_response_time:count",
-            "_response_ru_utime:count",
-            "_response_ru_stime:count",
-            "_response_db_txn_count:count",
-            "_response_db_txn_duration:count",
-        )
-    )
-)
-
-requests_counter = metrics.register_counter(
-    "requests_received",
-    labels=["method", "servlet", ],
-)
-
-outgoing_responses_counter = metrics.register_counter(
-    "responses",
-    labels=["method", "code"],
-)
-
-response_timer = metrics.register_counter(
-    "response_time_seconds",
-    labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_time:total",
-    ),
-)
-
-response_ru_utime = metrics.register_counter(
-    "response_ru_utime_seconds", labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_ru_utime:total",
-    ),
-)
-
-response_ru_stime = metrics.register_counter(
-    "response_ru_stime_seconds", labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_ru_stime:total",
-    ),
-)
-
-response_db_txn_count = metrics.register_counter(
-    "response_db_txn_count", labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_db_txn_count:total",
-    ),
-)
-
-# seconds spent waiting for db txns, excluding scheduling time, when processing
-# this request
-response_db_txn_duration = metrics.register_counter(
-    "response_db_txn_duration_seconds", labels=["method", "servlet", "tag"],
-    alternative_names=(
-        metrics.name_prefix + "_response_db_txn_duration:total",
-    ),
-)
-
-# seconds spent waiting for a db connection, when processing this request
-response_db_sched_duration = metrics.register_counter(
-    "response_db_sched_duration_seconds", labels=["method", "servlet", "tag"]
-)
-
-# size in bytes of the response written
-response_size = metrics.register_counter(
-    "response_size", labels=["method", "servlet", "tag"]
-)
-
-
-class RequestMetrics(object):
-    def start(self, time_msec, name):
-        self.start = time_msec
-        self.start_context = LoggingContext.current_context()
-        self.name = name
-
-    def stop(self, time_msec, request):
-        context = LoggingContext.current_context()
-
-        tag = ""
-        if context:
-            tag = context.tag
-
-            if context != self.start_context:
-                logger.warn(
-                    "Context have unexpectedly changed %r, %r",
-                    context, self.start_context
-                )
-                return
-
-        outgoing_responses_counter.inc(request.method, str(request.code))
-
-        response_count.inc(request.method, self.name, tag)
-
-        response_timer.inc_by(
-            time_msec - self.start, request.method,
-            self.name, tag
-        )
-
-        ru_utime, ru_stime = context.get_resource_usage()
-
-        response_ru_utime.inc_by(
-            ru_utime, request.method, self.name, tag
-        )
-        response_ru_stime.inc_by(
-            ru_stime, request.method, self.name, tag
-        )
-        response_db_txn_count.inc_by(
-            context.db_txn_count, request.method, self.name, tag
-        )
-        response_db_txn_duration.inc_by(
-            context.db_txn_duration_ms / 1000., request.method, self.name, tag
-        )
-        response_db_sched_duration.inc_by(
-            context.db_sched_duration_ms / 1000., request.method, self.name, tag
-        )
-
-        response_size.inc_by(request.sentLength, request.method, self.name, tag)

synapse/http/server.py

@@ -13,15 +13,11 @@
 # 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 cgi
-from six.moves import http_client
+
 
 from synapse.api.errors import (
     cs_exception, SynapseError, CodeMessageException, UnrecognizedRequestError, Codes
 )
-from synapse.http.request_metrics import (
-    requests_counter,
-)
 from synapse.util.logcontext import LoggingContext, PreserveLoggingContext
 from synapse.util.caches import intern_dict
 from synapse.util.metrics import Measure
@@ -45,174 +41,178 @@ import simplejson
 
 logger = logging.getLogger(__name__)
 
-HTML_ERROR_TEMPLATE = """<!DOCTYPE html>
-<html lang=en>
-  <head>
-    <meta charset="utf-8">
-    <title>Error {code}</title>
-  </head>
-  <body>
-     <p>{msg}</p>
-  </body>
-</html>
-"""
+metrics = synapse.metrics.get_metrics_for(__name__)
+
+# total number of responses served, split by method/servlet/tag
+response_count = metrics.register_counter(
+    "response_count",
+    labels=["method", "servlet", "tag"],
+    alternative_names=(
+        # the following are all deprecated aliases for the same metric
+        metrics.name_prefix + x for x in (
+            "_requests",
+            "_response_time:count",
+            "_response_ru_utime:count",
+            "_response_ru_stime:count",
+            "_response_db_txn_count:count",
+            "_response_db_txn_duration:count",
+        )
+    )
+)
+
+requests_counter = metrics.register_counter(
+    "requests_received",
+    labels=["method", "servlet", ],
+)
+
+outgoing_responses_counter = metrics.register_counter(
+    "responses",
+    labels=["method", "code"],
+)
+
+response_timer = metrics.register_counter(
+    "response_time_seconds",
+    labels=["method", "servlet", "tag"],
+    alternative_names=(
+        metrics.name_prefix + "_response_time:total",
+    ),
+)
+
+response_ru_utime = metrics.register_counter(
+    "response_ru_utime_seconds", labels=["method", "servlet", "tag"],
+    alternative_names=(
+        metrics.name_prefix + "_response_ru_utime:total",
+    ),
+)
+
+response_ru_stime = metrics.register_counter(
+    "response_ru_stime_seconds", labels=["method", "servlet", "tag"],
+    alternative_names=(
+        metrics.name_prefix + "_response_ru_stime:total",
+    ),
+)
+
+response_db_txn_count = metrics.register_counter(
+    "response_db_txn_count", labels=["method", "servlet", "tag"],
+    alternative_names=(
+        metrics.name_prefix + "_response_db_txn_count:total",
+    ),
+)
+
+# seconds spent waiting for db txns, excluding scheduling time, when processing
+# this request
+response_db_txn_duration = metrics.register_counter(
+    "response_db_txn_duration_seconds", labels=["method", "servlet", "tag"],
+    alternative_names=(
+        metrics.name_prefix + "_response_db_txn_duration:total",
+    ),
+)
+
+# seconds spent waiting for a db connection, when processing this request
+response_db_sched_duration = metrics.register_counter(
+    "response_db_sched_duration_seconds", labels=["method", "servlet", "tag"]
+)
+
+# size in bytes of the response written
+response_size = metrics.register_counter(
+    "response_size", labels=["method", "servlet", "tag"]
+)
+
+_next_request_id = 0
 
 
-def wrap_json_request_handler(h):
-    """Wraps a request handler method with exception handling.
+def request_handler(include_metrics=False):
+    """Decorator for ``wrap_request_handler``"""
+    return lambda request_handler: wrap_request_handler(request_handler, include_metrics)
 
-    Also adds logging as per wrap_request_handler_with_logging.
 
-    The handler method must have a signature of "handle_foo(self, request)",
-    where "self" must have a "clock" attribute (and "request" must be a
-    SynapseRequest).
+def wrap_request_handler(request_handler, include_metrics=False):
+    """Wraps a method that acts as a request handler with the necessary logging
+    and exception handling.
 
-    The handler must return a deferred. If the deferred succeeds we assume that
+    The method must have a signature of "handle_foo(self, request)". The
+    argument "self" must have "version_string" and "clock" attributes. The
+    argument "request" must be a twisted HTTP request.
+
+    The method must return a deferred. If the deferred succeeds we assume that
     a response has been sent. If the deferred fails with a SynapseError we use
     it to send a JSON response with the appropriate HTTP reponse code. If the
     deferred fails with any other type of error we send a 500 reponse.
+
+    We insert a unique request-id into the logging context for this request and
+    log the response and duration for this request.
     """
 
     @defer.inlineCallbacks
     def wrapped_request_handler(self, request):
-        try:
-            yield h(self, request)
-        except CodeMessageException as e:
-            code = e.code
-            if isinstance(e, SynapseError):
-                logger.info(
-                    "%s SynapseError: %s - %s", request, code, e.msg
-                )
-            else:
-                logger.exception(e)
-            respond_with_json(
-                request, code, cs_exception(e), send_cors=True,
-                pretty_print=_request_user_agent_is_curl(request),
-            )
+        global _next_request_id
+        request_id = "%s-%s" % (request.method, _next_request_id)
+        _next_request_id += 1
 
-        except Exception:
-            # failure.Failure() fishes the original Failure out
-            # of our stack, and thus gives us a sensible stack
-            # trace.
-            f = failure.Failure()
-            logger.error(
-                "Failed handle request via %r: %r: %s",
-                h,
-                request,
-                f.getTraceback().rstrip(),
-            )
-            respond_with_json(
-                request,
-                500,
-                {
-                    "error": "Internal server error",
-                    "errcode": Codes.UNKNOWN,
-                },
-                send_cors=True,
-                pretty_print=_request_user_agent_is_curl(request),
-            )
-
-    return wrap_request_handler_with_logging(wrapped_request_handler)
-
-
-def wrap_html_request_handler(h):
-    """Wraps a request handler method with exception handling.
-
-    Also adds logging as per wrap_request_handler_with_logging.
-
-    The handler method must have a signature of "handle_foo(self, request)",
-    where "self" must have a "clock" attribute (and "request" must be a
-    SynapseRequest).
-    """
-    def wrapped_request_handler(self, request):
-        d = defer.maybeDeferred(h, self, request)
-        d.addErrback(_return_html_error, request)
-        return d
-
-    return wrap_request_handler_with_logging(wrapped_request_handler)
-
-
-def _return_html_error(f, request):
-    """Sends an HTML error page corresponding to the given failure
-
-    Args:
-        f (twisted.python.failure.Failure):
-        request (twisted.web.iweb.IRequest):
-    """
-    if f.check(CodeMessageException):
-        cme = f.value
-        code = cme.code
-        msg = cme.msg
-
-        if isinstance(cme, SynapseError):
-            logger.info(
-                "%s SynapseError: %s - %s", request, code, msg
-            )
-        else:
-            logger.error(
-                "Failed handle request %r: %s",
-                request,
-                f.getTraceback().rstrip(),
-            )
-    else:
-        code = http_client.INTERNAL_SERVER_ERROR
-        msg = "Internal server error"
-
-        logger.error(
-            "Failed handle request %r: %s",
-            request,
-            f.getTraceback().rstrip(),
-        )
-
-    body = HTML_ERROR_TEMPLATE.format(
-        code=code, msg=cgi.escape(msg),
-    ).encode("utf-8")
-    request.setResponseCode(code)
-    request.setHeader(b"Content-Type", b"text/html; charset=utf-8")
-    request.setHeader(b"Content-Length", b"%i" % (len(body),))
-    request.write(body)
-    finish_request(request)
-
-
-def wrap_request_handler_with_logging(h):
-    """Wraps a request handler to provide logging and metrics
-
-    The handler method must have a signature of "handle_foo(self, request)",
-    where "self" must have a "clock" attribute (and "request" must be a
-    SynapseRequest).
-
-    As well as calling `request.processing` (which will log the response and
-    duration for this request), the wrapped request handler will insert the
-    request id into the logging context.
-    """
-    @defer.inlineCallbacks
-    def wrapped_request_handler(self, request):
-        """
-        Args:
-            self:
-            request (synapse.http.site.SynapseRequest):
-        """
-
-        request_id = request.get_request_id()
         with LoggingContext(request_id) as request_context:
-            request_context.request = request_id
             with Measure(self.clock, "wrapped_request_handler"):
+                request_metrics = RequestMetrics()
                 # we start the request metrics timer here with an initial stab
                 # at the servlet name. For most requests that name will be
                 # JsonResource (or a subclass), and JsonResource._async_render
                 # will update it once it picks a servlet.
                 servlet_name = self.__class__.__name__
-                with request.processing(servlet_name):
-                    with PreserveLoggingContext(request_context):
-                        d = defer.maybeDeferred(h, self, request)
+                request_metrics.start(self.clock, name=servlet_name)
 
-                        # record the arrival of the request *after*
-                        # dispatching to the handler, so that the handler
-                        # can update the servlet name in the request
-                        # metrics
-                        requests_counter.inc(request.method,
-                                             request.request_metrics.name)
-                        yield d
+                request_context.request = request_id
+                with request.processing():
+                    try:
+                        with PreserveLoggingContext(request_context):
+                            if include_metrics:
+                                yield request_handler(self, request, request_metrics)
+                            else:
+                                requests_counter.inc(request.method, servlet_name)
+                                yield request_handler(self, request)
+                    except CodeMessageException as e:
+                        code = e.code
+                        if isinstance(e, SynapseError):
+                            logger.info(
+                                "%s SynapseError: %s - %s", request, code, e.msg
+                            )
+                        else:
+                            logger.exception(e)
+                        outgoing_responses_counter.inc(request.method, str(code))
+                        respond_with_json(
+                            request, code, cs_exception(e), send_cors=True,
+                            pretty_print=_request_user_agent_is_curl(request),
+                            version_string=self.version_string,
+                        )
+                    except Exception:
+                        # failure.Failure() fishes the original Failure out
+                        # of our stack, and thus gives us a sensible stack
+                        # trace.
+                        f = failure.Failure()
+                        logger.error(
+                            "Failed handle request %s.%s on %r: %r: %s",
+                            request_handler.__module__,
+                            request_handler.__name__,
+                            self,
+                            request,
+                            f.getTraceback().rstrip(),
+                        )
+                        respond_with_json(
+                            request,
+                            500,
+                            {
+                                "error": "Internal server error",
+                                "errcode": Codes.UNKNOWN,
+                            },
+                            send_cors=True,
+                            pretty_print=_request_user_agent_is_curl(request),
+                            version_string=self.version_string,
+                        )
+                    finally:
+                        try:
+                            request_metrics.stop(
+                                self.clock, request
+                            )
+                        except Exception as e:
+                            logger.warn("Failed to stop metrics: %r", e)
     return wrapped_request_handler
 
 
@@ -262,6 +262,7 @@ class JsonResource(HttpServer, resource.Resource):
         self.canonical_json = canonical_json
         self.clock = hs.get_clock()
         self.path_regexs = {}
+        self.version_string = hs.version_string
         self.hs = hs
 
     def register_paths(self, method, path_patterns, callback):
@@ -277,9 +278,13 @@ class JsonResource(HttpServer, resource.Resource):
         self._async_render(request)
         return server.NOT_DONE_YET
 
-    @wrap_json_request_handler
+    # Disable metric reporting because _async_render does its own metrics.
+    # It does its own metric reporting because _async_render dispatches to
+    # a callback and it's the class name of that callback we want to report
+    # against rather than the JsonResource itself.
+    @request_handler(include_metrics=True)
     @defer.inlineCallbacks
-    def _async_render(self, request):
+    def _async_render(self, request, request_metrics):
         """ This gets called from render() every time someone sends us a request.
             This checks if anyone has registered a callback for that method and
             path.
@@ -291,7 +296,9 @@ class JsonResource(HttpServer, resource.Resource):
             servlet_classname = servlet_instance.__class__.__name__
         else:
             servlet_classname = "%r" % callback
-        request.request_metrics.name = servlet_classname
+
+        request_metrics.name = servlet_classname
+        requests_counter.inc(request.method, servlet_classname)
 
         # Now trigger the callback. If it returns a response, we send it
         # here. If it throws an exception, that is handled by the wrapper
@@ -338,12 +345,15 @@ class JsonResource(HttpServer, resource.Resource):
 
     def _send_response(self, request, code, response_json_object,
                        response_code_message=None):
+        outgoing_responses_counter.inc(request.method, str(code))
+
         # TODO: Only enable CORS for the requests that need it.
         respond_with_json(
             request, code, response_json_object,
             send_cors=True,
             response_code_message=response_code_message,
             pretty_print=_request_user_agent_is_curl(request),
+            version_string=self.version_string,
             canonical_json=self.canonical_json,
         )
 
@@ -376,6 +386,54 @@ def _unrecognised_request_handler(request):
     raise UnrecognizedRequestError()
 
 
+class RequestMetrics(object):
+    def start(self, clock, name):
+        self.start = clock.time_msec()
+        self.start_context = LoggingContext.current_context()
+        self.name = name
+
+    def stop(self, clock, request):
+        context = LoggingContext.current_context()
+
+        tag = ""
+        if context:
+            tag = context.tag
+
+            if context != self.start_context:
+                logger.warn(
+                    "Context have unexpectedly changed %r, %r",
+                    context, self.start_context
+                )
+                return
+
+        response_count.inc(request.method, self.name, tag)
+
+        response_timer.inc_by(
+            clock.time_msec() - self.start, request.method,
+            self.name, tag
+        )
+
+        ru_utime, ru_stime = context.get_resource_usage()
+
+        response_ru_utime.inc_by(
+            ru_utime, request.method, self.name, tag
+        )
+        response_ru_stime.inc_by(
+            ru_stime, request.method, self.name, tag
+        )
+        response_db_txn_count.inc_by(
+            context.db_txn_count, request.method, self.name, tag
+        )
+        response_db_txn_duration.inc_by(
+            context.db_txn_duration_ms / 1000., request.method, self.name, tag
+        )
+        response_db_sched_duration.inc_by(
+            context.db_sched_duration_ms / 1000., request.method, self.name, tag
+        )
+
+        response_size.inc_by(request.sentLength, request.method, self.name, tag)
+
+
 class RootRedirect(resource.Resource):
     """Redirects the root '/' path to another path."""
 
@@ -394,7 +452,7 @@ class RootRedirect(resource.Resource):
 
 def respond_with_json(request, code, json_object, send_cors=False,
                       response_code_message=None, pretty_print=False,
-                      canonical_json=True):
+                      version_string="", canonical_json=True):
     # could alternatively use request.notifyFinish() and flip a flag when
     # the Deferred fires, but since the flag is RIGHT THERE it seems like
     # a waste.
@@ -416,11 +474,12 @@ def respond_with_json(request, code, json_object, send_cors=False,
         request, code, json_bytes,
         send_cors=send_cors,
         response_code_message=response_code_message,
+        version_string=version_string
     )
 
 
 def respond_with_json_bytes(request, code, json_bytes, send_cors=False,
-                            response_code_message=None):
+                            version_string="", response_code_message=None):
     """Sends encoded JSON in response to the given request.
 
     Args:
@@ -434,6 +493,7 @@ def respond_with_json_bytes(request, code, json_bytes, send_cors=False,
 
     request.setResponseCode(code, message=response_code_message)
     request.setHeader(b"Content-Type", b"application/json")
+    request.setHeader(b"Server", version_string)
     request.setHeader(b"Content-Length", b"%d" % (len(json_bytes),))
     request.setHeader(b"Cache-Control", b"no-cache, no-store, must-revalidate")
 

synapse/http/site.py

@@ -12,48 +12,24 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from synapse.util.logcontext import LoggingContext
+from twisted.web.server import Site, Request
+
 import contextlib
 import logging
 import re
 import time
 
-from twisted.web.server import Site, Request
-
-from synapse.http.request_metrics import RequestMetrics
-from synapse.util.logcontext import LoggingContext
-
-logger = logging.getLogger(__name__)
-
 ACCESS_TOKEN_RE = re.compile(br'(\?.*access(_|%5[Ff])token=)[^&]*(.*)$')
 
-_next_request_seq = 0
-
 
 class SynapseRequest(Request):
-    """Class which encapsulates an HTTP request to synapse.
-
-    All of the requests processed in synapse are of this type.
-
-    It extends twisted's twisted.web.server.Request, and adds:
-     * Unique request ID
-     * Redaction of access_token query-params in __repr__
-     * Logging at start and end
-     * Metrics to record CPU, wallclock and DB time by endpoint.
-
-    It provides a method `processing` which should be called by the Resource
-    which is handling the request, and returns a context manager.
-
-    """
     def __init__(self, site, *args, **kw):
         Request.__init__(self, *args, **kw)
         self.site = site
         self.authenticated_entity = None
         self.start_time = 0
 
-        global _next_request_seq
-        self.request_seq = _next_request_seq
-        _next_request_seq += 1
-
     def __repr__(self):
         # We overwrite this so that we don't log ``access_token``
         return '<%s at 0x%x method=%s uri=%s clientproto=%s site=%s>' % (
@@ -65,9 +41,6 @@ class SynapseRequest(Request):
             self.site.site_tag,
         )
 
-    def get_request_id(self):
-        return "%s-%i" % (self.method, self.request_seq)
-
     def get_redacted_uri(self):
         return ACCESS_TOKEN_RE.sub(
             br'\1<redacted>\3',
@@ -77,16 +50,7 @@ class SynapseRequest(Request):
     def get_user_agent(self):
         return self.requestHeaders.getRawHeaders(b"User-Agent", [None])[-1]
 
-    def render(self, resrc):
-        # override the Server header which is set by twisted
-        self.setHeader("Server", self.site.server_version_string)
-        return Request.render(self, resrc)
-
-    def _started_processing(self, servlet_name):
-        self.start_time = int(time.time() * 1000)
-        self.request_metrics = RequestMetrics()
-        self.request_metrics.start(self.start_time, name=servlet_name)
-
+    def started_processing(self):
         self.site.access_logger.info(
             "%s - %s - Received request: %s %s",
             self.getClientIP(),
@@ -94,8 +58,10 @@ class SynapseRequest(Request):
             self.method,
             self.get_redacted_uri()
         )
+        self.start_time = int(time.time() * 1000)
+
+    def finished_processing(self):
 
-    def _finished_processing(self):
         try:
             context = LoggingContext.current_context()
             ru_utime, ru_stime = context.get_resource_usage()
@@ -106,8 +72,6 @@ class SynapseRequest(Request):
             ru_utime, ru_stime = (0, 0)
             db_txn_count, db_txn_duration_ms = (0, 0)
 
-        end_time = int(time.time() * 1000)
-
         self.site.access_logger.info(
             "%s - %s - {%s}"
             " Processed request: %dms (%dms, %dms) (%dms/%dms/%d)"
@@ -115,7 +79,7 @@ class SynapseRequest(Request):
             self.getClientIP(),
             self.site.site_tag,
             self.authenticated_entity,
-            end_time - self.start_time,
+            int(time.time() * 1000) - self.start_time,
             int(ru_utime * 1000),
             int(ru_stime * 1000),
             db_sched_duration_ms,
@@ -129,38 +93,11 @@ class SynapseRequest(Request):
             self.get_user_agent(),
         )
 
-        try:
-            self.request_metrics.stop(end_time, self)
-        except Exception as e:
-            logger.warn("Failed to stop metrics: %r", e)
-
     @contextlib.contextmanager
-    def processing(self, servlet_name):
-        """Record the fact that we are processing this request.
-
-        Returns a context manager; the correct way to use this is:
-
-        @defer.inlineCallbacks
-        def handle_request(request):
-            with request.processing("FooServlet"):
-                yield really_handle_the_request()
-
-        This will log the request's arrival. Once the context manager is
-        closed, the completion of the request will be logged, and the various
-        metrics will be updated.
-
-        Args:
-            servlet_name (str): the name of the servlet which will be
-                processing this request. This is used in the metrics.
-
-                It is possible to update this afterwards by updating
-                self.request_metrics.servlet_name.
-        """
-        # TODO: we should probably just move this into render() and finish(),
-        # to save having to call a separate method.
-        self._started_processing(servlet_name)
+    def processing(self):
+        self.started_processing()
         yield
-        self._finished_processing()
+        self.finished_processing()
 
 
 class XForwardedForRequest(SynapseRequest):
@@ -198,8 +135,7 @@ class SynapseSite(Site):
     Subclass of a twisted http Site that does access logging with python's
     standard logging
     """
-    def __init__(self, logger_name, site_tag, config, resource,
-                 server_version_string, *args, **kwargs):
+    def __init__(self, logger_name, site_tag, config, resource, *args, **kwargs):
         Site.__init__(self, resource, *args, **kwargs)
 
         self.site_tag = site_tag
@@ -207,7 +143,6 @@ class SynapseSite(Site):
         proxied = config.get("x_forwarded", False)
         self.requestFactory = SynapseRequestFactory(self, proxied)
         self.access_logger = logging.getLogger(logger_name)
-        self.server_version_string = server_version_string
 
     def log(self, request):
         pass

synapse/metrics/metric.py

@@ -71,8 +71,7 @@ class BaseMetric(object):
         """Render this metric for a single set of labels
 
         Args:
-            label_values (list[object]): values for each of the labels,
-                (which get stringified).
+            label_values (list[str]): values for each of the labels
             value: value of the metric at with these labels
 
         Returns:
@@ -325,4 +324,4 @@ def _escape_character(m):
 def _escape_label_value(value):
     """Takes a label value and escapes quotes, newlines and backslashes
     """
-    return re.sub(r"([\n\"\\])", _escape_character, str(value))
+    return re.sub(r"([\n\"\\])", _escape_character, value)

synapse/rest/client/v1/admin.py

@@ -151,11 +151,10 @@ class PurgeHistoryRestServlet(ClientV1RestServlet):
             if event.room_id != room_id:
                 raise SynapseError(400, "Event is for wrong room.")
 
-            token = yield self.store.get_topological_token_for_event(event_id)
-
+            depth = event.depth
             logger.info(
-                "[purge] purging up to token %s (event_id %s)",
-                token, event_id,
+                "[purge] purging up to depth %i (event_id %s)",
+                depth, event_id,
             )
         elif 'purge_up_to_ts' in body:
             ts = body['purge_up_to_ts']
@@ -175,9 +174,7 @@ class PurgeHistoryRestServlet(ClientV1RestServlet):
                 )
             )
             if room_event_after_stream_ordering:
-                token = yield self.store.get_topological_token_for_event(
-                    room_event_after_stream_ordering,
-                )
+                (_, depth, _) = room_event_after_stream_ordering
             else:
                 logger.warn(
                     "[purge] purging events not possible: No event found "
@@ -190,9 +187,9 @@ class PurgeHistoryRestServlet(ClientV1RestServlet):
                     errcode=Codes.NOT_FOUND,
                 )
             logger.info(
-                "[purge] purging up to token %d (received_ts %i => "
+                "[purge] purging up to depth %i (received_ts %i => "
                 "stream_ordering %i)",
-                token, ts, stream_ordering,
+                depth, ts, stream_ordering,
             )
         else:
             raise SynapseError(
@@ -202,7 +199,7 @@ class PurgeHistoryRestServlet(ClientV1RestServlet):
             )
 
         purge_id = yield self.handlers.message_handler.start_purge_history(
-            room_id, token,
+            room_id, depth,
             delete_local_events=delete_local_events,
         )
 
@@ -276,8 +273,8 @@ class ShutdownRoomRestServlet(ClientV1RestServlet):
     def __init__(self, hs):
         super(ShutdownRoomRestServlet, self).__init__(hs)
         self.store = hs.get_datastore()
+        self.handlers = hs.get_handlers()
         self.state = hs.get_state_handler()
-        self._room_creation_handler = hs.get_room_creation_handler()
         self.event_creation_handler = hs.get_event_creation_handler()
         self.room_member_handler = hs.get_room_member_handler()
 
@@ -299,7 +296,7 @@ class ShutdownRoomRestServlet(ClientV1RestServlet):
         message = content.get("message", self.DEFAULT_MESSAGE)
         room_name = content.get("room_name", "Content Violation Notification")
 
-        info = yield self._room_creation_handler.create_room(
+        info = yield self.handlers.room_creation_handler.create_room(
             room_creator_requester,
             config={
                 "preset": "public_chat",

synapse/rest/client/v1/pusher.py

@@ -176,6 +176,7 @@ class PushersRemoveRestServlet(RestServlet):
 
         request.setResponseCode(200)
         request.setHeader(b"Content-Type", b"text/html; charset=utf-8")
+        request.setHeader(b"Server", self.hs.version_string)
         request.setHeader(b"Content-Length", b"%d" % (
             len(PushersRemoveRestServlet.SUCCESS_HTML),
         ))

synapse/rest/client/v1/room.py

@@ -41,7 +41,7 @@ class RoomCreateRestServlet(ClientV1RestServlet):
 
     def __init__(self, hs):
         super(RoomCreateRestServlet, self).__init__(hs)
-        self._room_creation_handler = hs.get_room_creation_handler()
+        self.handlers = hs.get_handlers()
 
     def register(self, http_server):
         PATTERNS = "/createRoom"
@@ -64,7 +64,8 @@ class RoomCreateRestServlet(ClientV1RestServlet):
     def on_POST(self, request):
         requester = yield self.auth.get_user_by_req(request)
 
-        info = yield self._room_creation_handler.create_room(
+        handler = self.handlers.room_creation_handler
+        info = yield handler.create_room(
             requester, self.get_room_config(request)
         )
 

synapse/rest/client/v2_alpha/auth.py

@@ -129,6 +129,7 @@ class AuthRestServlet(RestServlet):
             html_bytes = html.encode("utf8")
             request.setResponseCode(200)
             request.setHeader(b"Content-Type", b"text/html; charset=utf-8")
+            request.setHeader(b"Server", self.hs.version_string)
             request.setHeader(b"Content-Length", b"%d" % (len(html_bytes),))
 
             request.write(html_bytes)
@@ -174,6 +175,7 @@ class AuthRestServlet(RestServlet):
             html_bytes = html.encode("utf8")
             request.setResponseCode(200)
             request.setHeader(b"Content-Type", b"text/html; charset=utf-8")
+            request.setHeader(b"Server", self.hs.version_string)
             request.setHeader(b"Content-Length", b"%d" % (len(html_bytes),))
 
             request.write(html_bytes)

synapse/rest/client/v2_alpha/notifications.py

@@ -88,7 +88,7 @@ class NotificationsServlet(RestServlet):
                     pa["topological_ordering"], pa["stream_ordering"]
                 )
             returned_push_actions.append(returned_pa)
-            next_token = str(pa["stream_ordering"])
+            next_token = pa["stream_ordering"]
 
         defer.returnValue((200, {
             "notifications": returned_push_actions,

synapse/rest/consent/consent_resource.py

@@ -1,210 +0,0 @@
-# -*- 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 hashlib import sha256
-import hmac
-import logging
-from os import path
-from six.moves import http_client
-
-import jinja2
-from jinja2 import TemplateNotFound
-from twisted.internet import defer
-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
-
-from synapse.api.errors import NotFoundError, SynapseError, StoreError
-from synapse.config import ConfigError
-from synapse.http.server import (
-    finish_request,
-    wrap_html_request_handler,
-)
-from synapse.http.servlet import parse_string
-from synapse.types import UserID
-
-
-# language to use for the templates. TODO: figure this out from Accept-Language
-TEMPLATE_LANGUAGE = "en"
-
-logger = logging.getLogger(__name__)
-
-# use hmac.compare_digest if we have it (python 2.7.7), else just use equality
-if hasattr(hmac, "compare_digest"):
-    compare_digest = hmac.compare_digest
-else:
-    def compare_digest(a, b):
-        return a == b
-
-
-class ConsentResource(Resource):
-    """A twisted Resource to display a privacy policy and gather consent to it
-
-    When accessed via GET, returns the privacy policy via a template.
-
-    When accessed via POST, records the user's consent in the database and
-    displays a success page.
-
-    The config should include a template_dir setting which contains templates
-    for the HTML. The directory should contain one subdirectory per language
-    (eg, 'en', 'fr'), and each language directory should contain the policy
-    document (named as '<version>.html') and a success page (success.html).
-
-    Both forms take a set of parameters from the browser. For the POST form,
-    these are normally sent as form parameters (but may be query-params); for
-    GET requests they must be query params. These are:
-
-        u: the complete mxid, or the localpart of the user giving their
-           consent. Required for both GET (where it is used as an input to the
-           template) and for POST (where it is used to find the row in the db
-           to update).
-
-        h: hmac_sha256(secret, u), where 'secret' is the privacy_secret in the
-           config file. If it doesn't match, the request is 403ed.
-
-        v: the version of the privacy policy being agreed to.
-
-           For GET: optional, and defaults to whatever was set in the config
-           file. Used to choose the version of the policy to pick from the
-           templates directory.
-
-           For POST: required; gives the value to be recorded in the database
-           against the user.
-    """
-    def __init__(self, hs):
-        """
-        Args:
-            hs (synapse.server.HomeServer): homeserver
-        """
-        Resource.__init__(self)
-
-        self.hs = hs
-        self.store = hs.get_datastore()
-
-        # this is required by the request_handler wrapper
-        self.clock = hs.get_clock()
-
-        consent_config = hs.config.consent_config
-        if consent_config is None:
-            raise ConfigError(
-                "Consent resource is enabled but user_consent section is "
-                "missing in config file.",
-            )
-
-        # daemonize changes the cwd to /, so make the path absolute now.
-        consent_template_directory = path.abspath(
-            consent_config["template_dir"],
-        )
-        if not path.isdir(consent_template_directory):
-            raise ConfigError(
-                "Could not find template directory '%s'" % (
-                    consent_template_directory,
-                ),
-            )
-
-        loader = jinja2.FileSystemLoader(consent_template_directory)
-        self._jinja_env = jinja2.Environment(loader=loader)
-
-        self._default_consent_verison = consent_config["default_version"]
-
-        if hs.config.form_secret is None:
-            raise ConfigError(
-                "Consent resource is enabled but form_secret is not set in "
-                "config file. It should be set to an arbitrary secret string.",
-            )
-
-        self._hmac_secret = hs.config.form_secret.encode("utf-8")
-
-    def render_GET(self, request):
-        self._async_render_GET(request)
-        return NOT_DONE_YET
-
-    @wrap_html_request_handler
-    def _async_render_GET(self, request):
-        """
-        Args:
-            request (twisted.web.http.Request):
-        """
-
-        version = parse_string(request, "v",
-                               default=self._default_consent_verison)
-        username = parse_string(request, "u", required=True)
-        userhmac = parse_string(request, "h", required=True)
-
-        self._check_hash(username, userhmac)
-
-        try:
-            self._render_template(
-                request, "%s.html" % (version,),
-                user=username, userhmac=userhmac, version=version,
-            )
-        except TemplateNotFound:
-            raise NotFoundError("Unknown policy version")
-
-    def render_POST(self, request):
-        self._async_render_POST(request)
-        return NOT_DONE_YET
-
-    @wrap_html_request_handler
-    @defer.inlineCallbacks
-    def _async_render_POST(self, request):
-        """
-        Args:
-            request (twisted.web.http.Request):
-        """
-        version = parse_string(request, "v", required=True)
-        username = parse_string(request, "u", required=True)
-        userhmac = parse_string(request, "h", required=True)
-
-        self._check_hash(username, userhmac)
-
-        if username.startswith('@'):
-            qualified_user_id = username
-        else:
-            qualified_user_id = UserID(username, self.hs.hostname).to_string()
-
-        try:
-            yield self.store.user_set_consent_version(qualified_user_id, version)
-        except StoreError as e:
-            if e.code != 404:
-                raise
-            raise NotFoundError("Unknown user")
-
-        try:
-            self._render_template(request, "success.html")
-        except TemplateNotFound:
-            raise NotFoundError("success.html not found")
-
-    def _render_template(self, request, template_name, **template_args):
-        # get_template checks for ".." so we don't need to worry too much
-        # about path traversal here.
-        template_html = self._jinja_env.get_template(
-            path.join(TEMPLATE_LANGUAGE, template_name)
-        )
-        html_bytes = template_html.render(**template_args).encode("utf8")
-
-        request.setHeader(b"Content-Type", b"text/html; charset=utf-8")
-        request.setHeader(b"Content-Length", b"%i" % len(html_bytes))
-        request.write(html_bytes)
-        finish_request(request)
-
-    def _check_hash(self, userid, userhmac):
-        want_mac = hmac.new(
-            key=self._hmac_secret,
-            msg=userid,
-            digestmod=sha256,
-        ).hexdigest()
-
-        if not compare_digest(want_mac, userhmac):
-            raise SynapseError(http_client.FORBIDDEN, "HMAC incorrect")

synapse/rest/key/v1/server_key_resource.py

@@ -49,6 +49,7 @@ class LocalKey(Resource):
     """
 
     def __init__(self, hs):
+        self.version_string = hs.version_string
         self.response_body = encode_canonical_json(
             self.response_json_object(hs.config)
         )
@@ -83,6 +84,7 @@ class LocalKey(Resource):
     def render_GET(self, request):
         return respond_with_json_bytes(
             request, 200, self.response_body,
+            version_string=self.version_string
         )
 
     def getChild(self, name, request):

synapse/rest/key/v2/local_key_resource.py

@@ -63,6 +63,7 @@ class LocalKey(Resource):
     isLeaf = True
 
     def __init__(self, hs):
+        self.version_string = hs.version_string
         self.config = hs.config
         self.clock = hs.clock
         self.update_response_body(self.clock.time_msec())
@@ -114,4 +115,5 @@ class LocalKey(Resource):
             self.update_response_body(time_now)
         return respond_with_json_bytes(
             request, 200, self.response_body,
+            version_string=self.version_string
         )

synapse/rest/key/v2/remote_key_resource.py

@@ -12,9 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from synapse.http.server import (
-    respond_with_json_bytes, wrap_json_request_handler,
-)
+from synapse.http.server import request_handler, respond_with_json_bytes
 from synapse.http.servlet import parse_integer, parse_json_object_from_request
 from synapse.api.errors import SynapseError, Codes
 from synapse.crypto.keyring import KeyLookupError
@@ -93,6 +91,7 @@ class RemoteKey(Resource):
     def __init__(self, hs):
         self.keyring = hs.get_keyring()
         self.store = hs.get_datastore()
+        self.version_string = hs.version_string
         self.clock = hs.get_clock()
         self.federation_domain_whitelist = hs.config.federation_domain_whitelist
 
@@ -100,7 +99,7 @@ class RemoteKey(Resource):
         self.async_render_GET(request)
         return NOT_DONE_YET
 
-    @wrap_json_request_handler
+    @request_handler()
     @defer.inlineCallbacks
     def async_render_GET(self, request):
         if len(request.postpath) == 1:
@@ -125,7 +124,7 @@ class RemoteKey(Resource):
         self.async_render_POST(request)
         return NOT_DONE_YET
 
-    @wrap_json_request_handler
+    @request_handler()
     @defer.inlineCallbacks
     def async_render_POST(self, request):
         content = parse_json_object_from_request(request)
@@ -241,4 +240,5 @@ class RemoteKey(Resource):
 
             respond_with_json_bytes(
                 request, 200, result_io.getvalue(),
+                version_string=self.version_string
             )

synapse/rest/media/v1/download_resource.py

@@ -12,18 +12,16 @@
 # 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 twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
-
-from synapse.http.server import (
-    set_cors_headers,
-    wrap_json_request_handler,
-)
 import synapse.http.servlet
+
 from ._base import parse_media_id, respond_404
+from twisted.web.resource import Resource
+from synapse.http.server import request_handler, set_cors_headers
+
+from twisted.web.server import NOT_DONE_YET
+from twisted.internet import defer
+
+import logging
 
 logger = logging.getLogger(__name__)
 
@@ -37,14 +35,15 @@ class DownloadResource(Resource):
         self.media_repo = media_repo
         self.server_name = hs.hostname
 
-        # this is expected by @wrap_json_request_handler
+        # Both of these are expected by @request_handler()
         self.clock = hs.get_clock()
+        self.version_string = hs.version_string
 
     def render_GET(self, request):
         self._async_render_GET(request)
         return NOT_DONE_YET
 
-    @wrap_json_request_handler
+    @request_handler()
     @defer.inlineCallbacks
     def _async_render_GET(self, request):
         set_cors_headers(request)

synapse/rest/media/v1/preview_url_resource.py

@@ -40,9 +40,8 @@ from synapse.util.stringutils import random_string
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.http.client import SpiderHttpClient
 from synapse.http.server import (
-    respond_with_json_bytes,
+    request_handler, respond_with_json_bytes,
     respond_with_json,
-    wrap_json_request_handler,
 )
 from synapse.util.async import ObservableDeferred
 from synapse.util.stringutils import is_ascii
@@ -58,6 +57,7 @@ class PreviewUrlResource(Resource):
 
         self.auth = hs.get_auth()
         self.clock = hs.get_clock()
+        self.version_string = hs.version_string
         self.filepaths = media_repo.filepaths
         self.max_spider_size = hs.config.max_spider_size
         self.server_name = hs.hostname
@@ -90,7 +90,7 @@ class PreviewUrlResource(Resource):
         self._async_render_GET(request)
         return NOT_DONE_YET
 
-    @wrap_json_request_handler
+    @request_handler()
     @defer.inlineCallbacks
     def _async_render_GET(self, request):
 

synapse/rest/media/v1/thumbnail_resource.py

@@ -14,21 +14,18 @@
 # limitations under the License.
 
 
-import logging
-
-from twisted.internet import defer
-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
-
-from synapse.http.server import (
-    set_cors_headers,
-    wrap_json_request_handler,
-)
-from synapse.http.servlet import parse_integer, parse_string
 from ._base import (
-    FileInfo, parse_media_id, respond_404, respond_with_file,
+    parse_media_id, respond_404, respond_with_file, FileInfo,
     respond_with_responder,
 )
+from twisted.web.resource import Resource
+from synapse.http.servlet import parse_string, parse_integer
+from synapse.http.server import request_handler, set_cors_headers
+
+from twisted.web.server import NOT_DONE_YET
+from twisted.internet import defer
+
+import logging
 
 logger = logging.getLogger(__name__)
 
@@ -44,13 +41,14 @@ class ThumbnailResource(Resource):
         self.media_storage = media_storage
         self.dynamic_thumbnails = hs.config.dynamic_thumbnails
         self.server_name = hs.hostname
+        self.version_string = hs.version_string
         self.clock = hs.get_clock()
 
     def render_GET(self, request):
         self._async_render_GET(request)
         return NOT_DONE_YET
 
-    @wrap_json_request_handler
+    @request_handler()
     @defer.inlineCallbacks
     def _async_render_GET(self, request):
         set_cors_headers(request)

synapse/rest/media/v1/upload_resource.py

@@ -13,17 +13,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
-
-from twisted.internet import defer
-from twisted.web.resource import Resource
-from twisted.web.server import NOT_DONE_YET
+from synapse.http.server import respond_with_json, request_handler
 
 from synapse.api.errors import SynapseError
-from synapse.http.server import (
-    respond_with_json,
-    wrap_json_request_handler,
-)
+
+from twisted.web.server import NOT_DONE_YET
+from twisted.internet import defer
+
+from twisted.web.resource import Resource
+
+import logging
 
 logger = logging.getLogger(__name__)
 
@@ -41,6 +40,7 @@ class UploadResource(Resource):
         self.server_name = hs.hostname
         self.auth = hs.get_auth()
         self.max_upload_size = hs.config.max_upload_size
+        self.version_string = hs.version_string
         self.clock = hs.get_clock()
 
     def render_POST(self, request):
@@ -51,7 +51,7 @@ class UploadResource(Resource):
         respond_with_json(request, 200, {}, send_cors=True)
         return NOT_DONE_YET
 
-    @wrap_json_request_handler
+    @request_handler()
     @defer.inlineCallbacks
     def _async_render_POST(self, request):
         requester = yield self.auth.get_user_by_req(request)

synapse/server.py

@@ -46,7 +46,6 @@ from synapse.handlers.devicemessage import DeviceMessageHandler
 from synapse.handlers.device import DeviceHandler
 from synapse.handlers.e2e_keys import E2eKeysHandler
 from synapse.handlers.presence import PresenceHandler
-from synapse.handlers.room import RoomCreationHandler
 from synapse.handlers.room_list import RoomListHandler
 from synapse.handlers.room_member import RoomMemberMasterHandler
 from synapse.handlers.room_member_worker import RoomMemberWorkerHandler
@@ -72,7 +71,6 @@ from synapse.rest.media.v1.media_repository import (
     MediaRepository,
     MediaRepositoryResource,
 )
-from synapse.server_notices.server_notices_manager import ServerNoticesManager
 from synapse.state import StateHandler, StateResolutionHandler
 from synapse.storage import DataStore
 from synapse.streams.events import EventSources
@@ -99,9 +97,6 @@ class HomeServer(object):
     which must be implemented by the subclass. This code may call any of the
     required "get" methods on the instance to obtain the sub-dependencies that
     one requires.
-
-    Attributes:
-        config (synapse.config.homeserver.HomeserverConfig):
     """
 
     DEPENDENCIES = [
@@ -111,7 +106,6 @@ class HomeServer(object):
         'federation_server',
         'handlers',
         'auth',
-        'room_creation_handler',
         'state_handler',
         'state_resolution_handler',
         'presence_handler',
@@ -157,7 +151,6 @@ class HomeServer(object):
         'spam_checker',
         'room_member_handler',
         'federation_registry',
-        'server_notices_manager',
     ]
 
     def __init__(self, hostname, **kwargs):
@@ -231,9 +224,6 @@ class HomeServer(object):
     def build_simple_http_client(self):
         return SimpleHttpClient(self)
 
-    def build_room_creation_handler(self):
-        return RoomCreationHandler(self)
-
     def build_state_handler(self):
         return StateHandler(self)
 
@@ -400,9 +390,6 @@ class HomeServer(object):
     def build_federation_registry(self):
         return FederationHandlerRegistry()
 
-    def build_server_notices_manager(self):
-        return ServerNoticesManager(self)
-
     def remove_pusher(self, app_id, push_key, user_id):
         return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
 

synapse/server.pyi

@@ -1,5 +1,4 @@
 import synapse.api.auth
-import synapse.config.homeserver
 import synapse.federation.transaction_queue
 import synapse.federation.transport.client
 import synapse.handlers
@@ -9,16 +8,11 @@ import synapse.handlers.device
 import synapse.handlers.e2e_keys
 import synapse.handlers.set_password
 import synapse.rest.media.v1.media_repository
-import synapse.server_notices.server_notices_manager
 import synapse.state
 import synapse.storage
 
 
 class HomeServer(object):
-    @property
-    def config(self) -> synapse.config.homeserver.HomeServerConfig:
-        pass
-
     def get_auth(self) -> synapse.api.auth.Auth:
         pass
 
@@ -46,12 +40,6 @@ class HomeServer(object):
     def get_deactivate_account_handler(self) -> synapse.handlers.deactivate_account.DeactivateAccountHandler:
         pass
 
-    def get_room_creation_handler(self) -> synapse.handlers.room.RoomCreationHandler:
-        pass
-
-    def get_event_creation_handler(self) -> synapse.handlers.message.EventCreationHandler:
-        pass
-
     def get_set_password_handler(self) -> synapse.handlers.set_password.SetPasswordHandler:
         pass
 
@@ -66,6 +54,3 @@ class HomeServer(object):
 
     def get_media_repository(self) -> synapse.rest.media.v1.media_repository.MediaRepository:
         pass
-
-    def get_server_notices_manager(self) -> synapse.server_notices.server_notices_manager.ServerNoticesManager:
-        pass

synapse/server_notices/server_notices_manager.py

@@ -1,131 +0,0 @@
-# -*- 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, Membership, RoomCreationPreset
-from synapse.types import create_requester
-from synapse.util.caches.descriptors import cachedInlineCallbacks
-
-logger = logging.getLogger(__name__)
-
-
-class ServerNoticesManager(object):
-    def __init__(self, hs):
-        """
-
-        Args:
-            hs (synapse.server.HomeServer):
-        """
-
-        self._store = hs.get_datastore()
-        self._config = hs.config
-        self._room_creation_handler = hs.get_room_creation_handler()
-        self._event_creation_handler = hs.get_event_creation_handler()
-
-    def is_enabled(self):
-        """Checks if server notices are enabled on this server.
-
-        Returns:
-            bool
-        """
-        return self._config.server_notices_mxid is not None
-
-    @defer.inlineCallbacks
-    def send_notice(self, user_id, event_content):
-        """Send a notice to the given user
-
-        Creates the server notices room, if none exists.
-
-        Args:
-            user_id (str): mxid of user to send event to.
-            event_content (dict): content of event to send
-
-        Returns:
-            Deferrred[None]
-        """
-        room_id = yield self.get_notice_room_for_user(user_id)
-
-        system_mxid = self._config.server_notices_mxid
-        requester = create_requester(system_mxid)
-
-        logger.info("Sending server notice to %s", user_id)
-
-        yield self._event_creation_handler.create_and_send_nonmember_event(
-            requester, {
-                "type": EventTypes.Message,
-                "room_id": room_id,
-                "sender": system_mxid,
-                "content": event_content,
-            },
-            ratelimit=False,
-        )
-
-    @cachedInlineCallbacks()
-    def get_notice_room_for_user(self, user_id):
-        """Get the room for notices for a given user
-
-        If we have not yet created a notice room for this user, create it
-
-        Args:
-            user_id (str): complete user id for the user we want a room for
-
-        Returns:
-            str: room id of notice room.
-        """
-        if not self.is_enabled():
-            raise Exception("Server notices not enabled")
-
-        rooms = yield self._store.get_rooms_for_user_where_membership_is(
-            user_id, [Membership.INVITE, Membership.JOIN],
-        )
-        system_mxid = self._config.server_notices_mxid
-        for room in rooms:
-            # it's worth noting that there is an asymmetry here in that we
-            # expect the user to be invited or joined, but the system user must
-            # be joined. This is kinda deliberate, in that if somebody somehow
-            # manages to invite the system user to a room, that doesn't make it
-            # the server notices room.
-            user_ids = yield self._store.get_users_in_room(room.room_id)
-            if system_mxid in user_ids:
-                # we found a room which our user shares with the system notice
-                # user
-                logger.info("Using room %s", room.room_id)
-                defer.returnValue(room.room_id)
-
-        # apparently no existing notice room: create a new one
-        logger.info("Creating server notices room for %s", user_id)
-
-        requester = create_requester(system_mxid)
-        info = yield self._room_creation_handler.create_room(
-            requester,
-            config={
-                "preset": RoomCreationPreset.PRIVATE_CHAT,
-                "name": self._config.server_notices_room_name,
-                "power_level_content_override": {
-                    "users_default": -10,
-                },
-                "invite": (user_id,)
-            },
-            ratelimit=False,
-            creator_join_profile={
-                "displayname": self._config.server_notices_mxid_display_name,
-            },
-        )
-        room_id = info['room_id']
-
-        logger.info("Created server notices room %s for %s", room_id, user_id)
-        defer.returnValue(room_id)

synapse/storage/__init__.py

@@ -14,11 +14,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import datetime
-from dateutil import tz
-import time
-import logging
-
 from synapse.storage.devices import DeviceStore
 from .appservice import (
     ApplicationServiceStore, ApplicationServiceTransactionStore
@@ -60,6 +55,10 @@ from .engines import PostgresEngine
 from synapse.api.constants import PresenceState
 from synapse.util.caches.stream_change_cache import StreamChangeCache
 
+
+import logging
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -214,9 +213,6 @@ class DataStore(RoomMemberStore, RoomStore,
         self._stream_order_on_start = self.get_room_max_stream_ordering()
         self._min_stream_order_on_start = self.get_room_min_stream_ordering()
 
-        # Used in _generate_user_daily_visits to keep track of progress
-        self._last_user_visit_update = self._get_start_of_day()
-
         super(DataStore, self).__init__(db_conn, hs)
 
     def take_presence_startup_info(self):
@@ -351,65 +347,6 @@ class DataStore(RoomMemberStore, RoomStore,
 
         return self.runInteraction("count_r30_users", _count_r30_users)
 
-    def _get_start_of_day(self):
-        """
-        Returns millisecond unixtime for start of UTC day.
-        """
-        now = datetime.datetime.utcnow()
-        today_start = datetime.datetime(now.year, now.month,
-                                        now.day, tzinfo=tz.tzutc())
-        return int(time.mktime(today_start.timetuple())) * 1000
-
-    def generate_user_daily_visits(self):
-        """
-        Generates daily visit data for use in cohort/ retention analysis
-        """
-        def _generate_user_daily_visits(txn):
-            logger.info("Calling _generate_user_daily_visits")
-            today_start = self._get_start_of_day()
-            a_day_in_milliseconds = 24 * 60 * 60 * 1000
-            now = self.clock.time_msec()
-
-            sql = """
-                INSERT INTO user_daily_visits (user_id, device_id, timestamp)
-                    SELECT u.user_id, u.device_id, ?
-                    FROM user_ips AS u
-                    LEFT JOIN (
-                      SELECT user_id, device_id, timestamp FROM user_daily_visits
-                      WHERE timestamp = ?
-                    ) udv
-                    ON u.user_id = udv.user_id AND u.device_id=udv.device_id
-                    WHERE last_seen > ? AND last_seen <= ? AND udv.timestamp IS NULL
-            """
-
-            # This means that the day has rolled over but there could still
-            # be entries from the previous day. There is an edge case
-            # where if the user logs in at 23:59 and overwrites their
-            # last_seen at 00:01 then they will not be counted in the
-            # previous day's stats - it is important that the query is run
-            # often to minimise this case.
-            if today_start > self._last_user_visit_update:
-                yesterday_start = today_start - a_day_in_milliseconds
-                txn.execute(sql, (
-                    yesterday_start, yesterday_start,
-                    self._last_user_visit_update, today_start
-                ))
-                self._last_user_visit_update = today_start
-
-            txn.execute(sql, (
-                today_start, today_start,
-                self._last_user_visit_update,
-                now
-            ))
-            # Update _last_user_visit_update to now. The reason to do this
-            # rather just clamping to the beginning of the day is to limit
-            # the size of the join - meaning that the query can be run more
-            # frequently
-            self._last_user_visit_update = now
-
-        return self.runInteraction("generate_user_daily_visits",
-                                   _generate_user_daily_visits)
-
     def get_users(self):
         """Function to reterive a list of users in users table.
 

synapse/storage/client_ips.py

@@ -55,13 +55,6 @@ class ClientIpStore(background_updates.BackgroundUpdateStore):
             columns=["user_id", "last_seen"],
         )
 
-        self.register_background_index_update(
-            "user_ips_last_seen_only_index",
-            index_name="user_ips_last_seen_only",
-            table="user_ips",
-            columns=["last_seen"],
-        )
-
         # (user_id, access_token, ip) -> (user_agent, device_id, last_seen)
         self._batch_row_update = {}
 

synapse/storage/engines/__init__.py

@@ -15,7 +15,7 @@
 
 from ._base import IncorrectDatabaseSetup
 from .postgres import PostgresEngine
-from .sqlite3 import Sqlite3Engine
+from .sqlite3_engine import Sqlite3Engine
 
 import importlib
 import platform

synapse/storage/engines/sqlite3_engine.py

@@ -15,6 +15,7 @@
 
 from synapse.storage.prepare_database import prepare_database
 
+import sqlite3
 import struct
 import threading
 
@@ -25,6 +26,11 @@ class Sqlite3Engine(object):
     def __init__(self, database_module, database_config):
         self.module = database_module
 
+        if sqlite3.sqlite_version_info < (3, 15, 0):
+            raise RuntimeError(
+                "SQLite3 version is too old, Synapse requires 3.15 or later",
+            )
+
         # The current max state_group, or None if we haven't looked
         # in the DB yet.
         self._current_state_group_id = None

synapse/storage/event_push_actions.py

@@ -18,6 +18,8 @@ from synapse.storage._base import SQLBaseStore, LoggingTransaction
 from twisted.internet import defer
 from synapse.util.async import sleep
 from synapse.util.caches.descriptors import cachedInlineCallbacks
+from synapse.types import RoomStreamToken
+from .stream import lower_bound
 
 import logging
 import simplejson as json
@@ -97,7 +99,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
     def _get_unread_counts_by_receipt_txn(self, txn, room_id, user_id,
                                           last_read_event_id):
         sql = (
-            "SELECT stream_ordering"
+            "SELECT stream_ordering, topological_ordering"
             " FROM events"
             " WHERE room_id = ? AND event_id = ?"
         )
@@ -109,12 +111,17 @@ class EventPushActionsWorkerStore(SQLBaseStore):
             return {"notify_count": 0, "highlight_count": 0}
 
         stream_ordering = results[0][0]
+        topological_ordering = results[0][1]
 
         return self._get_unread_counts_by_pos_txn(
-            txn, room_id, user_id, stream_ordering
+            txn, room_id, user_id, topological_ordering, stream_ordering
         )
 
-    def _get_unread_counts_by_pos_txn(self, txn, room_id, user_id, stream_ordering):
+    def _get_unread_counts_by_pos_txn(self, txn, room_id, user_id, topological_ordering,
+                                      stream_ordering):
+        token = RoomStreamToken(
+            topological_ordering, stream_ordering
+        )
 
         # First get number of notifications.
         # We don't need to put a notif=1 clause as all rows always have
@@ -125,10 +132,10 @@ class EventPushActionsWorkerStore(SQLBaseStore):
             " WHERE"
             " user_id = ?"
             " AND room_id = ?"
-            " AND stream_ordering > ?"
-        )
+            " AND %s"
+        ) % (lower_bound(token, self.database_engine, inclusive=False),)
 
-        txn.execute(sql, (user_id, room_id, stream_ordering))
+        txn.execute(sql, (user_id, room_id))
         row = txn.fetchone()
         notify_count = row[0] if row else 0
 
@@ -148,10 +155,10 @@ class EventPushActionsWorkerStore(SQLBaseStore):
             " highlight = 1"
             " AND user_id = ?"
             " AND room_id = ?"
-            " AND stream_ordering > ?"
-        )
+            " AND %s"
+        ) % (lower_bound(token, self.database_engine, inclusive=False),)
 
-        txn.execute(sql, (user_id, room_id, stream_ordering))
+        txn.execute(sql, (user_id, room_id))
         row = txn.fetchone()
         highlight_count = row[0] if row else 0
 
@@ -202,6 +209,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 "   ep.highlight "
                 " FROM ("
                 "   SELECT room_id,"
+                "       MAX(topological_ordering) as topological_ordering,"
                 "       MAX(stream_ordering) as stream_ordering"
                 "   FROM events"
                 "   INNER JOIN receipts_linearized USING (room_id, event_id)"
@@ -211,7 +219,13 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 " event_push_actions AS ep"
                 " WHERE"
                 "   ep.room_id = rl.room_id"
-                "   AND ep.stream_ordering > rl.stream_ordering"
+                "   AND ("
+                "       ep.topological_ordering > rl.topological_ordering"
+                "       OR ("
+                "           ep.topological_ordering = rl.topological_ordering"
+                "           AND ep.stream_ordering > rl.stream_ordering"
+                "       )"
+                "   )"
                 "   AND ep.user_id = ?"
                 "   AND ep.stream_ordering > ?"
                 "   AND ep.stream_ordering <= ?"
@@ -304,6 +318,7 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 "  ep.highlight, e.received_ts"
                 " FROM ("
                 "   SELECT room_id,"
+                "       MAX(topological_ordering) as topological_ordering,"
                 "       MAX(stream_ordering) as stream_ordering"
                 "   FROM events"
                 "   INNER JOIN receipts_linearized USING (room_id, event_id)"
@@ -314,7 +329,13 @@ class EventPushActionsWorkerStore(SQLBaseStore):
                 " INNER JOIN events AS e USING (room_id, event_id)"
                 " WHERE"
                 "   ep.room_id = rl.room_id"
-                "   AND ep.stream_ordering > rl.stream_ordering"
+                "   AND ("
+                "       ep.topological_ordering > rl.topological_ordering"
+                "       OR ("
+                "           ep.topological_ordering = rl.topological_ordering"
+                "           AND ep.stream_ordering > rl.stream_ordering"
+                "       )"
+                "   )"
                 "   AND ep.user_id = ?"
                 "   AND ep.stream_ordering > ?"
                 "   AND ep.stream_ordering <= ?"
@@ -741,10 +762,10 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
         )
 
     def _remove_old_push_actions_before_txn(self, txn, room_id, user_id,
-                                            stream_ordering):
+                                            topological_ordering, stream_ordering):
         """
         Purges old push actions for a user and room before a given
-        stream_ordering.
+        topological_ordering.
 
         We however keep a months worth of highlighted notifications, so that
         users can still get a list of recent highlights.
@@ -753,7 +774,7 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
             txn: The transcation
             room_id: Room ID to delete from
             user_id: user ID to delete for
-            stream_ordering: The lowest stream ordering which will
+            topological_ordering: The lowest topological ordering which will
                                   not be deleted.
         """
         txn.call_after(
@@ -772,9 +793,9 @@ class EventPushActionsStore(EventPushActionsWorkerStore):
         txn.execute(
             "DELETE FROM event_push_actions "
             " WHERE user_id = ? AND room_id = ? AND "
-            " stream_ordering <= ?"
+            " topological_ordering <= ?"
             " AND ((stream_ordering < ? AND highlight = 1) or highlight = 0)",
-            (user_id, room_id, stream_ordering, self.stream_ordering_month_ago)
+            (user_id, room_id, topological_ordering, self.stream_ordering_month_ago)
         )
 
         txn.execute("""

synapse/storage/events.py

@@ -33,7 +33,7 @@ from synapse.util.metrics import Measure
 from synapse.api.constants import EventTypes
 from synapse.api.errors import SynapseError
 from synapse.util.caches.descriptors import cached, cachedInlineCallbacks
-from synapse.types import get_domain_from_id, RoomStreamToken
+from synapse.types import get_domain_from_id
 import synapse.metrics
 
 # these are only included to make the type annotations work
@@ -1803,14 +1803,15 @@ class EventsStore(EventsWorkerStore):
         return self.runInteraction("get_all_new_events", get_all_new_events_txn)
 
     def purge_history(
-        self, room_id, token, delete_local_events,
+        self, room_id, topological_ordering, delete_local_events,
     ):
         """Deletes room history before a certain point
 
         Args:
             room_id (str):
 
-            token (str): A topological token to delete events before
+            topological_ordering (int):
+                minimum topo ordering to preserve
 
             delete_local_events (bool):
                 if True, we will delete local events as well as remote ones
@@ -1820,15 +1821,13 @@ class EventsStore(EventsWorkerStore):
 
         return self.runInteraction(
             "purge_history",
-            self._purge_history_txn, room_id, token,
+            self._purge_history_txn, room_id, topological_ordering,
             delete_local_events,
         )
 
     def _purge_history_txn(
-        self, txn, room_id, token_str, delete_local_events,
+        self, txn, room_id, topological_ordering, delete_local_events,
     ):
-        token = RoomStreamToken.parse(token_str)
-
         # Tables that should be pruned:
         #     event_auth
         #     event_backward_extremities
@@ -1873,13 +1872,6 @@ class EventsStore(EventsWorkerStore):
             " ON events_to_purge(should_delete)",
         )
 
-        # We do joins against events_to_purge for e.g. calculating state
-        # groups to purge, etc., so lets make an index.
-        txn.execute(
-            "CREATE INDEX events_to_purge_id"
-            " ON events_to_purge(event_id)",
-        )
-
         # First ensure that we're not about to delete all the forward extremeties
         txn.execute(
             "SELECT e.event_id, e.depth FROM events as e "
@@ -1892,7 +1884,7 @@ class EventsStore(EventsWorkerStore):
         rows = txn.fetchall()
         max_depth = max(row[0] for row in rows)
 
-        if max_depth <= token.topological:
+        if max_depth <= topological_ordering:
             # We need to ensure we don't delete all the events from the datanase
             # otherwise we wouldn't be able to send any events (due to not
             # having any backwards extremeties)
@@ -1908,7 +1900,7 @@ class EventsStore(EventsWorkerStore):
             should_delete_expr += " AND event_id NOT LIKE ?"
             should_delete_params += ("%:" + self.hs.hostname, )
 
-        should_delete_params += (room_id, token.topological)
+        should_delete_params += (room_id, topological_ordering)
 
         txn.execute(
             "INSERT INTO events_to_purge"
@@ -1931,13 +1923,13 @@ class EventsStore(EventsWorkerStore):
         logger.info("[purge] Finding new backward extremities")
 
         # We calculate the new entries for the backward extremeties by finding
-        # events to be purged that are pointed to by events we're not going to
-        # purge.
+        # all events that point to events that are to be purged
         txn.execute(
             "SELECT DISTINCT e.event_id FROM events_to_purge AS e"
             " INNER JOIN event_edges AS ed ON e.event_id = ed.prev_event_id"
-            " LEFT JOIN events_to_purge AS ep2 ON ed.event_id = ep2.event_id"
-            " WHERE ep2.event_id IS NULL",
+            " INNER JOIN events AS e2 ON e2.event_id = ed.event_id"
+            " WHERE e2.topological_ordering >= ?",
+            (topological_ordering, )
         )
         new_backwards_extrems = txn.fetchall()
 
@@ -1961,22 +1953,16 @@ class EventsStore(EventsWorkerStore):
 
         # Get all state groups that are only referenced by events that are
         # to be deleted.
-        # This works by first getting state groups that we may want to delete,
-        # joining against event_to_state_groups to get events that use that
-        # state group, then left joining against events_to_purge again. Any
-        # state group where the left join produce *no nulls* are referenced
-        # only by events that are going to be purged.
-        txn.execute("""
-            SELECT state_group FROM
-            (
-                SELECT DISTINCT state_group FROM events_to_purge
-                INNER JOIN event_to_state_groups USING (event_id)
-            ) AS sp
-            INNER JOIN event_to_state_groups USING (state_group)
-            LEFT JOIN events_to_purge AS ep USING (event_id)
-            GROUP BY state_group
-            HAVING SUM(CASE WHEN ep.event_id IS NULL THEN 1 ELSE 0 END) = 0
-        """)
+        txn.execute(
+            "SELECT state_group FROM event_to_state_groups"
+            " INNER JOIN events USING (event_id)"
+            " WHERE state_group IN ("
+            "   SELECT DISTINCT state_group FROM events_to_purge"
+            "   INNER JOIN event_to_state_groups USING (event_id)"
+            " )"
+            " GROUP BY state_group HAVING MAX(topological_ordering) < ?",
+            (topological_ordering, )
+        )
 
         state_rows = txn.fetchall()
         logger.info("[purge] found %i redundant state groups", len(state_rows))
@@ -2123,25 +2109,10 @@ class EventsStore(EventsWorkerStore):
         #
         # So, let's stick it at the end so that we don't block event
         # persistence.
-        #
-        # We do this by calculating the minimum depth of the backwards
-        # extremities. However, the events in event_backward_extremities
-        # are ones we don't have yet so we need to look at the events that
-        # point to it via event_edges table.
-        txn.execute("""
-            SELECT COALESCE(MIN(depth), 0)
-            FROM event_backward_extremities AS eb
-            INNER JOIN event_edges AS eg ON eg.prev_event_id = eb.event_id
-            INNER JOIN events AS e ON e.event_id = eg.event_id
-            WHERE eb.room_id = ?
-        """, (room_id,))
-        min_depth, = txn.fetchone()
-
-        logger.info("[purge] updating room_depth to %d", min_depth)
-
+        logger.info("[purge] updating room_depth")
         txn.execute(
             "UPDATE room_depth SET min_depth = ? WHERE room_id = ?",
-            (min_depth, room_id,)
+            (topological_ordering, room_id,)
         )
 
         # finally, drop the temp table. this will commit the txn in sqlite,

synapse/storage/prepare_database.py

@@ -26,7 +26,7 @@ logger = logging.getLogger(__name__)
 
 # Remember to update this number every time a change is made to database
 # schema files, so the users will be informed on server restarts.
-SCHEMA_VERSION = 49
+SCHEMA_VERSION = 48
 
 dir_path = os.path.abspath(os.path.dirname(__file__))
 

synapse/storage/receipts.py

@@ -297,22 +297,18 @@ class ReceiptsWorkerStore(SQLBaseStore):
         if receipt_type != "m.read":
             return
 
-        # Returns either an ObservableDeferred or the raw result
+        # Returns an ObservableDeferred
         res = self.get_users_with_read_receipts_in_room.cache.get(
             room_id, None, update_metrics=False,
         )
 
-        # first handle the Deferred case
-        if isinstance(res, defer.Deferred):
-            if res.called:
+        if res:
+            if isinstance(res, defer.Deferred) and res.called:
                 res = res.result
-            else:
-                res = None
-
-        if res and user_id in res:
-            # We'd only be adding to the set, so no point invalidating if the
-            # user is already there
-            return
+            if user_id in res:
+                # We'd only be adding to the set, so no point invalidating if the
+                # user is already there
+                return
 
         self.get_users_with_read_receipts_in_room.invalidate((room_id,))
 
@@ -411,6 +407,7 @@ class ReceiptsStore(ReceiptsWorkerStore):
                 txn,
                 room_id=room_id,
                 user_id=user_id,
+                topological_ordering=topological_ordering,
                 stream_ordering=stream_ordering,
             )
 

synapse/storage/registration.py

@@ -286,24 +286,6 @@ class RegistrationStore(RegistrationWorkerStore,
             "user_set_password_hash", user_set_password_hash_txn
         )
 
-    def user_set_consent_version(self, user_id, consent_version):
-        """Updates the user table to record privacy policy consent
-
-        Args:
-            user_id (str): full mxid of the user to update
-            consent_version (str): version of the policy the user has consented
-                to
-
-        Raises:
-            StoreError(404) if user not found
-        """
-        return self._simple_update_one(
-            table='users',
-            keyvalues={'name': user_id, },
-            updatevalues={'consent_version': consent_version, },
-            desc="user_set_consent_version"
-        )
-
     def user_delete_access_tokens(self, user_id, except_token_id=None,
                                   device_id=None):
         """
@@ -544,42 +526,3 @@ class RegistrationStore(RegistrationWorkerStore,
         except self.database_engine.module.IntegrityError:
             ret = yield self.get_3pid_guest_access_token(medium, address)
             defer.returnValue(ret)
-
-    def add_user_pending_deactivation(self, user_id):
-        """
-        Adds a user to the table of users who need to be parted from all the rooms they're
-        in
-        """
-        return self._simple_insert(
-            "users_pending_deactivation",
-            values={
-                "user_id": user_id,
-            },
-            desc="add_user_pending_deactivation",
-        )
-
-    def del_user_pending_deactivation(self, user_id):
-        """
-        Removes the given user to the table of users who need to be parted from all the
-        rooms they're in, effectively marking that user as fully deactivated.
-        """
-        return self._simple_delete_one(
-            "users_pending_deactivation",
-            keyvalues={
-                "user_id": user_id,
-            },
-            desc="del_user_pending_deactivation",
-        )
-
-    def get_user_pending_deactivation(self):
-        """
-        Gets one user from the table of users waiting to be parted from all the rooms
-        they're in.
-        """
-        return self._simple_select_one_onecol(
-            "users_pending_deactivation",
-            keyvalues={},
-            retcol="user_id",
-            allow_none=True,
-            desc="get_users_pending_deactivation",
-        )

synapse/storage/schema/delta/48/add_user_consent.sql

@@ -1,18 +0,0 @@
-/* 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.
- */
-
-/* record the version of the privacy policy the user has consented to
- */
-ALTER TABLE users ADD COLUMN consent_version TEXT;

synapse/storage/schema/delta/48/deactivated_users.sql

@@ -1,25 +0,0 @@
-/* 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.
- */
-
-/*
- * Store any accounts that have been requested to be deactivated.
- * We part the account from all the rooms its in when its
- * deactivated. This can take some time and synapse may be restarted
- * before it completes, so store the user IDs here until the process
- * is complete.
- */
-CREATE TABLE users_pending_deactivation (
-    user_id TEXT NOT NULL
-);

synapse/storage/schema/delta/49/add_user_daily_visits.sql

@@ -1,21 +0,0 @@
-/* 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 user_daily_visits ( user_id TEXT NOT NULL,
-                                 device_id TEXT,
-                                 timestamp BIGINT NOT NULL );
-CREATE INDEX user_daily_visits_uts_idx ON user_daily_visits(user_id, timestamp);
-CREATE INDEX user_daily_visits_ts_idx ON user_daily_visits(timestamp);

synapse/storage/schema/delta/49/add_user_ips_last_seen_only_index.sql

@@ -1,17 +0,0 @@
-/* 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.
- */
-
-INSERT into background_updates (update_name, progress_json)
-    VALUES ('user_ips_last_seen_only_index', '{}');

synapse/storage/stream.py

@@ -38,16 +38,16 @@ from twisted.internet import defer
 from synapse.storage._base import SQLBaseStore
 from synapse.storage.events import EventsWorkerStore
 
+from synapse.util.caches.descriptors import cached
 from synapse.types import RoomStreamToken
 from synapse.util.caches.stream_change_cache import StreamChangeCache
 from synapse.util.logcontext import make_deferred_yieldable, run_in_background
-from synapse.storage.engines import PostgresEngine
+from synapse.storage.engines import PostgresEngine, Sqlite3Engine
 
 import abc
 import logging
 
 from six.moves import range
-from collections import namedtuple
 
 
 logger = logging.getLogger(__name__)
@@ -60,12 +60,6 @@ _STREAM_TOKEN = "stream"
 _TOPOLOGICAL_TOKEN = "topological"
 
 
-# Used as return values for pagination APIs
-_EventDictReturn = namedtuple("_EventDictReturn", (
-    "event_id", "topological_ordering", "stream_ordering",
-))
-
-
 def lower_bound(token, engine, inclusive=False):
     inclusive = "=" if inclusive else ""
     if token.topological is None:
@@ -233,55 +227,54 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
     @defer.inlineCallbacks
     def get_room_events_stream_for_room(self, room_id, from_key, to_key, limit=0,
                                         order='DESC'):
+        # Note: If from_key is None then we return in topological order. This
+        # is because in that case we're using this as a "get the last few messages
+        # in a room" function, rather than "get new messages since last sync"
+        if from_key is not None:
+            from_id = RoomStreamToken.parse_stream_token(from_key).stream
+        else:
+            from_id = None
+        to_id = RoomStreamToken.parse_stream_token(to_key).stream
 
-        """Get new room events in stream ordering since `from_key`.
-
-        Args:
-            room_id (str)
-            from_key (str): Token from which no events are returned before
-            to_key (str): Token from which no events are returned after. (This
-                is typically the current stream token)
-            limit (int): Maximum number of events to return
-            order (str): Either "DESC" or "ASC". Determines which events are
-                returned when the result is limited. If "DESC" then the most
-                recent `limit` events are returned, otherwise returns the
-                oldest `limit` events.
-
-        Returns:
-            Deferred[tuple[list[FrozenEvent], str]]: Returns the list of
-            events (in ascending order) and the token from the start of
-            the chunk of events returned.
-        """
         if from_key == to_key:
             defer.returnValue(([], from_key))
 
-        from_id = RoomStreamToken.parse_stream_token(from_key).stream
-        to_id = RoomStreamToken.parse_stream_token(to_key).stream
+        if from_id:
+            has_changed = yield self._events_stream_cache.has_entity_changed(
+                room_id, from_id
+            )
 
-        has_changed = yield self._events_stream_cache.has_entity_changed(
-            room_id, from_id
-        )
-
-        if not has_changed:
-            defer.returnValue(([], from_key))
+            if not has_changed:
+                defer.returnValue(([], from_key))
 
         def f(txn):
-            sql = (
-                "SELECT event_id, stream_ordering FROM events WHERE"
-                " room_id = ?"
-                " AND not outlier"
-                " AND stream_ordering > ? AND stream_ordering <= ?"
-                " ORDER BY stream_ordering %s LIMIT ?"
-            ) % (order,)
-            txn.execute(sql, (room_id, from_id, to_id, limit))
+            if from_id is not None:
+                sql = (
+                    "SELECT event_id, stream_ordering FROM events WHERE"
+                    " room_id = ?"
+                    " AND not outlier"
+                    " AND stream_ordering > ? AND stream_ordering <= ?"
+                    " ORDER BY stream_ordering %s LIMIT ?"
+                ) % (order,)
+                txn.execute(sql, (room_id, from_id, to_id, limit))
+            else:
+                sql = (
+                    "SELECT event_id, stream_ordering FROM events WHERE"
+                    " room_id = ?"
+                    " AND not outlier"
+                    " AND stream_ordering <= ?"
+                    " ORDER BY topological_ordering %s, stream_ordering %s LIMIT ?"
+                ) % (order, order,)
+                txn.execute(sql, (room_id, to_id, limit))
+
+            rows = self.cursor_to_dict(txn)
 
-            rows = [_EventDictReturn(row[0], None, row[1]) for row in txn]
             return rows
 
         rows = yield self.runInteraction("get_room_events_stream_for_room", f)
 
         ret = yield self._get_events(
-            [r.event_id for r in rows],
+            [r["event_id"] for r in rows],
             get_prev_content=True
         )
 
@@ -291,7 +284,7 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             ret.reverse()
 
         if rows:
-            key = "s%d" % min(r.stream_ordering for r in rows)
+            key = "s%d" % min(r["stream_ordering"] for r in rows)
         else:
             # Assume we didn't get anything because there was nothing to
             # get.
@@ -301,7 +294,10 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
 
     @defer.inlineCallbacks
     def get_membership_changes_for_user(self, user_id, from_key, to_key):
-        from_id = RoomStreamToken.parse_stream_token(from_key).stream
+        if from_key is not None:
+            from_id = RoomStreamToken.parse_stream_token(from_key).stream
+        else:
+            from_id = None
         to_id = RoomStreamToken.parse_stream_token(to_key).stream
 
         if from_key == to_key:
@@ -315,24 +311,34 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
                 defer.returnValue([])
 
         def f(txn):
-            sql = (
-                "SELECT m.event_id, stream_ordering FROM events AS e,"
-                " room_memberships AS m"
-                " WHERE e.event_id = m.event_id"
-                " AND m.user_id = ?"
-                " AND e.stream_ordering > ? AND e.stream_ordering <= ?"
-                " ORDER BY e.stream_ordering ASC"
-            )
-            txn.execute(sql, (user_id, from_id, to_id,))
-
-            rows = [_EventDictReturn(row[0], None, row[1]) for row in txn]
+            if from_id is not None:
+                sql = (
+                    "SELECT m.event_id, stream_ordering FROM events AS e,"
+                    " room_memberships AS m"
+                    " WHERE e.event_id = m.event_id"
+                    " AND m.user_id = ?"
+                    " AND e.stream_ordering > ? AND e.stream_ordering <= ?"
+                    " ORDER BY e.stream_ordering ASC"
+                )
+                txn.execute(sql, (user_id, from_id, to_id,))
+            else:
+                sql = (
+                    "SELECT m.event_id, stream_ordering FROM events AS e,"
+                    " room_memberships AS m"
+                    " WHERE e.event_id = m.event_id"
+                    " AND m.user_id = ?"
+                    " AND stream_ordering <= ?"
+                    " ORDER BY stream_ordering ASC"
+                )
+                txn.execute(sql, (user_id, to_id,))
+            rows = self.cursor_to_dict(txn)
 
             return rows
 
         rows = yield self.runInteraction("get_membership_changes_for_user", f)
 
         ret = yield self._get_events(
-            [r.event_id for r in rows],
+            [r["event_id"] for r in rows],
             get_prev_content=True
         )
 
@@ -341,28 +347,14 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         defer.returnValue(ret)
 
     @defer.inlineCallbacks
-    def get_recent_events_for_room(self, room_id, limit, end_token):
-        """Get the most recent events in the room in topological ordering.
-
-        Args:
-            room_id (str)
-            limit (int)
-            end_token (str): The stream token representing now.
-
-        Returns:
-            Deferred[tuple[list[FrozenEvent],  str]]: Returns a list of
-            events and a token pointing to the start of the returned
-            events.
-            The events returned are in ascending order.
-        """
-
+    def get_recent_events_for_room(self, room_id, limit, end_token, from_token=None):
         rows, token = yield self.get_recent_event_ids_for_room(
-            room_id, limit, end_token,
+            room_id, limit, end_token, from_token
         )
 
         logger.debug("stream before")
         events = yield self._get_events(
-            [r.event_id for r in rows],
+            [r["event_id"] for r in rows],
             get_prev_content=True
         )
         logger.debug("stream after")
@@ -371,37 +363,61 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
 
         defer.returnValue((events, token))
 
-    @defer.inlineCallbacks
-    def get_recent_event_ids_for_room(self, room_id, limit, end_token):
-        """Get the most recent events in the room in topological ordering.
+    @cached(num_args=4)
+    def get_recent_event_ids_for_room(self, room_id, limit, end_token, from_token=None):
+        end_token = RoomStreamToken.parse_stream_token(end_token)
 
-        Args:
-            room_id (str)
-            limit (int)
-            end_token (str): The stream token representing now.
+        if from_token is None:
+            sql = (
+                "SELECT stream_ordering, topological_ordering, event_id"
+                " FROM events"
+                " WHERE room_id = ? AND stream_ordering <= ? AND outlier = ?"
+                " ORDER BY topological_ordering DESC, stream_ordering DESC"
+                " LIMIT ?"
+            )
+        else:
+            from_token = RoomStreamToken.parse_stream_token(from_token)
+            sql = (
+                "SELECT stream_ordering, topological_ordering, event_id"
+                " FROM events"
+                " WHERE room_id = ? AND stream_ordering > ?"
+                " AND stream_ordering <= ? AND outlier = ?"
+                " ORDER BY topological_ordering DESC, stream_ordering DESC"
+                " LIMIT ?"
+            )
 
-        Returns:
-            Deferred[tuple[list[_EventDictReturn],  str]]: Returns a list of
-            _EventDictReturn and a token pointing to the start of the returned
-            events.
-            The events returned are in ascending order.
-        """
-        # Allow a zero limit here, and no-op.
-        if limit == 0:
-            defer.returnValue(([], end_token))
+        def get_recent_events_for_room_txn(txn):
+            if from_token is None:
+                txn.execute(sql, (room_id, end_token.stream, False, limit,))
+            else:
+                txn.execute(sql, (
+                    room_id, from_token.stream, end_token.stream, False, limit
+                ))
 
-        end_token = RoomStreamToken.parse(end_token)
+            rows = self.cursor_to_dict(txn)
 
-        rows, token = yield self.runInteraction(
-            "get_recent_event_ids_for_room", self._paginate_room_events_txn,
-            room_id, from_token=end_token, limit=limit,
+            rows.reverse()  # As we selected with reverse ordering
+
+            if rows:
+                # Tokens are positions between events.
+                # This token points *after* the last event in the chunk.
+                # We need it to point to the event before it in the chunk
+                # since we are going backwards so we subtract one from the
+                # stream part.
+                topo = rows[0]["topological_ordering"]
+                toke = rows[0]["stream_ordering"] - 1
+                start_token = str(RoomStreamToken(topo, toke))
+
+                token = (start_token, str(end_token))
+            else:
+                token = (str(end_token), str(end_token))
+
+            return rows, token
+
+        return self.runInteraction(
+            "get_recent_events_for_room", get_recent_events_for_room_txn
         )
 
-        # We want to return the results in ascending order.
-        rows.reverse()
-
-        defer.returnValue((rows, token))
-
     def get_room_event_after_stream_ordering(self, room_id, stream_ordering):
         """Gets details of the first event in a room at or after a stream ordering
 
@@ -504,20 +520,10 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
 
     @staticmethod
     def _set_before_and_after(events, rows, topo_order=True):
-        """Inserts ordering information to events' internal metadata from
-        the DB rows.
-
-        Args:
-            events (list[FrozenEvent])
-            rows (list[_EventDictReturn])
-            topo_order (bool): Whether the events were ordered topologically
-                or by stream ordering. If true then all rows should have a non
-                null topological_ordering.
-        """
         for event, row in zip(events, rows):
-            stream = row.stream_ordering
-            if topo_order and row.topological_ordering:
-                topo = row.topological_ordering
+            stream = row["stream_ordering"]
+            if topo_order:
+                topo = event.depth
             else:
                 topo = None
             internal = event.internal_metadata
@@ -589,27 +595,87 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             retcols=["stream_ordering", "topological_ordering"],
         )
 
-        # Paginating backwards includes the event at the token, but paginating
-        # forward doesn't.
-        before_token = RoomStreamToken(
-            results["topological_ordering"] - 1,
-            results["stream_ordering"],
-        )
-
-        after_token = RoomStreamToken(
+        token = RoomStreamToken(
             results["topological_ordering"],
             results["stream_ordering"],
         )
 
-        rows, start_token = self._paginate_room_events_txn(
-            txn, room_id, before_token, direction='b', limit=before_limit,
-        )
-        events_before = [r.event_id for r in rows]
+        if isinstance(self.database_engine, Sqlite3Engine):
+            # SQLite3 doesn't optimise ``(x < a) OR (x = a AND y < b)``
+            # So we give pass it to SQLite3 as the UNION ALL of the two queries.
 
-        rows, end_token = self._paginate_room_events_txn(
-            txn, room_id, after_token, direction='f', limit=after_limit,
-        )
-        events_after = [r.event_id for r in rows]
+            query_before = (
+                "SELECT topological_ordering, stream_ordering, event_id FROM events"
+                " WHERE room_id = ? AND topological_ordering < ?"
+                " UNION ALL"
+                " SELECT topological_ordering, stream_ordering, event_id FROM events"
+                " WHERE room_id = ? AND topological_ordering = ? AND stream_ordering < ?"
+                " ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ?"
+            )
+            before_args = (
+                room_id, token.topological,
+                room_id, token.topological, token.stream,
+                before_limit,
+            )
+
+            query_after = (
+                "SELECT topological_ordering, stream_ordering, event_id FROM events"
+                " WHERE room_id = ? AND topological_ordering > ?"
+                " UNION ALL"
+                " SELECT topological_ordering, stream_ordering, event_id FROM events"
+                " WHERE room_id = ? AND topological_ordering = ? AND stream_ordering > ?"
+                " ORDER BY topological_ordering ASC, stream_ordering ASC LIMIT ?"
+            )
+            after_args = (
+                room_id, token.topological,
+                room_id, token.topological, token.stream,
+                after_limit,
+            )
+        else:
+            query_before = (
+                "SELECT topological_ordering, stream_ordering, event_id FROM events"
+                " WHERE room_id = ? AND %s"
+                " ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ?"
+            ) % (upper_bound(token, self.database_engine, inclusive=False),)
+
+            before_args = (room_id, before_limit)
+
+            query_after = (
+                "SELECT topological_ordering, stream_ordering, event_id FROM events"
+                " WHERE room_id = ? AND %s"
+                " ORDER BY topological_ordering ASC, stream_ordering ASC LIMIT ?"
+            ) % (lower_bound(token, self.database_engine, inclusive=False),)
+
+            after_args = (room_id, after_limit)
+
+        txn.execute(query_before, before_args)
+
+        rows = self.cursor_to_dict(txn)
+        events_before = [r["event_id"] for r in rows]
+
+        if rows:
+            start_token = str(RoomStreamToken(
+                rows[0]["topological_ordering"],
+                rows[0]["stream_ordering"] - 1,
+            ))
+        else:
+            start_token = str(RoomStreamToken(
+                token.topological,
+                token.stream - 1,
+            ))
+
+        txn.execute(query_after, after_args)
+
+        rows = self.cursor_to_dict(txn)
+        events_after = [r["event_id"] for r in rows]
+
+        if rows:
+            end_token = str(RoomStreamToken(
+                rows[-1]["topological_ordering"],
+                rows[-1]["stream_ordering"],
+            ))
+        else:
+            end_token = str(token)
 
         return {
             "before": {
@@ -672,28 +738,17 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
     def has_room_changed_since(self, room_id, stream_id):
         return self._events_stream_cache.has_entity_changed(room_id, stream_id)
 
-    def _paginate_room_events_txn(self, txn, room_id, from_token, to_token=None,
-                                  direction='b', limit=-1, event_filter=None):
-        """Returns list of events before or after a given token.
 
-        Args:
-            txn
-            room_id (str)
-            from_token (RoomStreamToken): The token used to stream from
-            to_token (RoomStreamToken|None): A token which if given limits the
-                results to only those before
-            direction(char): Either 'b' or 'f' to indicate whether we are
-                paginating forwards or backwards from `from_key`.
-            limit (int): The maximum number of events to return. Zero or less
-                means no limit.
-            event_filter (Filter|None): If provided filters the events to
-                those that match the filter.
+class StreamStore(StreamWorkerStore):
+    def get_room_max_stream_ordering(self):
+        return self._stream_id_gen.get_current_token()
 
-        Returns:
-            Deferred[tuple[list[_EventDictReturn], str]]: Returns the results
-            as a list of _EventDictReturn and a token that points to the end
-            of the result set.
-        """
+    def get_room_min_stream_ordering(self):
+        return self._backfill_id_gen.get_current_token()
+
+    @defer.inlineCallbacks
+    def paginate_room_events(self, room_id, from_key, to_key=None,
+                             direction='b', limit=-1, event_filter=None):
         # Tokens really represent positions between elements, but we use
         # the convention of pointing to the event before the gap. Hence
         # we have a bit of asymmetry when it comes to equalities.
@@ -701,20 +756,20 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         if direction == 'b':
             order = "DESC"
             bounds = upper_bound(
-                from_token, self.database_engine
+                RoomStreamToken.parse(from_key), self.database_engine
             )
-            if to_token:
+            if to_key:
                 bounds = "%s AND %s" % (bounds, lower_bound(
-                    to_token, self.database_engine
+                    RoomStreamToken.parse(to_key), self.database_engine
                 ))
         else:
             order = "ASC"
             bounds = lower_bound(
-                from_token, self.database_engine
+                RoomStreamToken.parse(from_key), self.database_engine
             )
-            if to_token:
+            if to_key:
                 bounds = "%s AND %s" % (bounds, upper_bound(
-                    to_token, self.database_engine
+                    RoomStreamToken.parse(to_key), self.database_engine
                 ))
 
         filter_clause, filter_args = filter_to_clause(event_filter)
@@ -730,8 +785,7 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             limit_str = ""
 
         sql = (
-            "SELECT event_id, topological_ordering, stream_ordering"
-            " FROM events"
+            "SELECT * FROM events"
             " WHERE outlier = ? AND room_id = ? AND %(bounds)s"
             " ORDER BY topological_ordering %(order)s,"
             " stream_ordering %(order)s %(limit)s"
@@ -741,72 +795,35 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             "limit": limit_str
         }
 
-        txn.execute(sql, args)
+        def f(txn):
+            txn.execute(sql, args)
 
-        rows = [_EventDictReturn(row[0], row[1], row[2]) for row in txn]
+            rows = self.cursor_to_dict(txn)
 
-        if rows:
-            topo = rows[-1].topological_ordering
-            toke = rows[-1].stream_ordering
-            if direction == 'b':
-                # Tokens are positions between events.
-                # This token points *after* the last event in the chunk.
-                # We need it to point to the event before it in the chunk
-                # when we are going backwards so we subtract one from the
-                # stream part.
-                toke -= 1
-            next_token = RoomStreamToken(topo, toke)
-        else:
-            # TODO (erikj): We should work out what to do here instead.
-            next_token = to_token if to_token else from_token
+            if rows:
+                topo = rows[-1]["topological_ordering"]
+                toke = rows[-1]["stream_ordering"]
+                if direction == 'b':
+                    # Tokens are positions between events.
+                    # This token points *after* the last event in the chunk.
+                    # We need it to point to the event before it in the chunk
+                    # when we are going backwards so we subtract one from the
+                    # stream part.
+                    toke -= 1
+                next_token = str(RoomStreamToken(topo, toke))
+            else:
+                # TODO (erikj): We should work out what to do here instead.
+                next_token = to_key if to_key else from_key
 
-        return rows, str(next_token),
+            return rows, next_token,
 
-    @defer.inlineCallbacks
-    def paginate_room_events(self, room_id, from_key, to_key=None,
-                             direction='b', limit=-1, event_filter=None):
-        """Returns list of events before or after a given token.
-
-        Args:
-            room_id (str)
-            from_key (str): The token used to stream from
-            to_key (str|None): A token which if given limits the results to
-                only those before
-            direction(char): Either 'b' or 'f' to indicate whether we are
-                paginating forwards or backwards from `from_key`.
-            limit (int): The maximum number of events to return. Zero or less
-                means no limit.
-            event_filter (Filter|None): If provided filters the events to
-                those that match the filter.
-
-        Returns:
-            tuple[list[dict], str]: Returns the results as a list of dicts and
-            a token that points to the end of the result set. The dicts have
-            the keys "event_id", "topological_ordering" and "stream_orderign".
-        """
-
-        from_key = RoomStreamToken.parse(from_key)
-        if to_key:
-            to_key = RoomStreamToken.parse(to_key)
-
-        rows, token = yield self.runInteraction(
-            "paginate_room_events", self._paginate_room_events_txn,
-            room_id, from_key, to_key, direction, limit, event_filter,
-        )
+        rows, token = yield self.runInteraction("paginate_room_events", f)
 
         events = yield self._get_events(
-            [r.event_id for r in rows],
+            [r["event_id"] for r in rows],
             get_prev_content=True
         )
 
         self._set_before_and_after(events, rows)
 
         defer.returnValue((events, token))
-
-
-class StreamStore(StreamWorkerStore):
-    def get_room_max_stream_ordering(self):
-        return self._stream_id_gen.get_current_token()
-
-    def get_room_min_stream_ordering(self):
-        return self._backfill_id_gen.get_current_token()

tests/storage/test_event_push_actions.py

@@ -55,7 +55,7 @@ class EventPushActionsStoreTestCase(tests.unittest.TestCase):
         def _assert_counts(noitf_count, highlight_count):
             counts = yield self.store.runInteraction(
                 "", self.store._get_unread_counts_by_pos_txn,
-                room_id, user_id, 0
+                room_id, user_id, 0, 0
             )
             self.assertEquals(
                 counts,
@@ -86,7 +86,7 @@ class EventPushActionsStoreTestCase(tests.unittest.TestCase):
         def _mark_read(stream, depth):
             return self.store.runInteraction(
                 "", self.store._remove_old_push_actions_before_txn,
-                room_id, user_id, stream
+                room_id, user_id, depth, stream
             )
 
         yield _assert_counts(0, 0)