start.py: Support SYNAPSE_CONFIG_TEMPLATE_DIR

* Store generated worker config files in SYNAPSE_CONFIG_DIR, rather than hardcoding
  `/conf`.
* Move the `workers_have_been_configured` filepath to the config dir.

CHANGES.md

@@ -1,57 +1,3 @@
-# Synapse 1.117.0 (2024-10-15)
-
-No significant changes since 1.117.0rc1.
-
-
-
-
-# Synapse 1.117.0rc1 (2024-10-08)
-
-### Features
-
-- Add config option `redis.password_path`. ([\#17717](https://github.com/element-hq/synapse/issues/17717))
-
-### Bugfixes
-
-- Fix a rare bug introduced in v1.29.0 where invalidating a user's access token from a worker could raise an error. ([\#17779](https://github.com/element-hq/synapse/issues/17779))
-- In the response to `GET /_matrix/client/versions`, set the `unstable_features` flag for [MSC4140](https://github.com/matrix-org/matrix-spec-proposals/pull/4140) to `false` when server configuration disables support for delayed events. ([\#17780](https://github.com/element-hq/synapse/issues/17780))
-- Improve input validation and room membership checks in admin redaction API. ([\#17792](https://github.com/element-hq/synapse/issues/17792))
-
-### Improved Documentation
-
-- Clarify the docstring of `test_forget_when_not_left`. ([\#17628](https://github.com/element-hq/synapse/issues/17628))
-- Add documentation note about PYTHONMALLOC for accurate jemalloc memory tracking. Contributed by @hensg. ([\#17709](https://github.com/element-hq/synapse/issues/17709))
-- Remove spurious "TODO UPDATE ALL THIS" note in the Debian installation docs. ([\#17749](https://github.com/element-hq/synapse/issues/17749))
-- Explain how load balancing works for `federation_sender_instances`. ([\#17776](https://github.com/element-hq/synapse/issues/17776))
-
-### Internal Changes
-
-- Minor performance increase for large accounts using sliding sync. ([\#17751](https://github.com/element-hq/synapse/issues/17751))
-- Increase performance of the notifier when there are many syncing users. ([\#17765](https://github.com/element-hq/synapse/issues/17765), [\#17766](https://github.com/element-hq/synapse/issues/17766))
-- Fix performance of streams that don't change often. ([\#17767](https://github.com/element-hq/synapse/issues/17767))
-- Improve performance of sliding sync connections that do not ask for any rooms. ([\#17768](https://github.com/element-hq/synapse/issues/17768))
-- Reduce overhead of sliding sync E2EE loops. ([\#17771](https://github.com/element-hq/synapse/issues/17771))
-- Sliding sync minor performance speed up using new table. ([\#17787](https://github.com/element-hq/synapse/issues/17787))
-- Sliding sync minor performance improvement by omitting unchanged data from incremental responses. ([\#17788](https://github.com/element-hq/synapse/issues/17788))
-- Speed up sliding sync when there are many active subscriptions. ([\#17789](https://github.com/element-hq/synapse/issues/17789))
-- Add missing license headers on new source files. ([\#17799](https://github.com/element-hq/synapse/issues/17799))
-
-
-
-### Updates to locked dependencies
-
-* Bump phonenumbers from 8.13.45 to 8.13.46. ([\#17773](https://github.com/element-hq/synapse/issues/17773))
-* Bump python-multipart from 0.0.10 to 0.0.12. ([\#17772](https://github.com/element-hq/synapse/issues/17772))
-* Bump regex from 1.10.6 to 1.11.0. ([\#17770](https://github.com/element-hq/synapse/issues/17770))
-* Bump ruff from 0.6.7 to 0.6.8. ([\#17774](https://github.com/element-hq/synapse/issues/17774))
-
-# Synapse 1.116.0 (2024-10-01)
-
-No significant changes since 1.116.0rc2.
-
-
-
-
 # Synapse 1.116.0rc2 (2024-09-26)
 
 ### Features

Cargo.lock

@@ -444,9 +444,9 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.11.0"
+version = "1.10.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "38200e5ee88914975b69f657f0801b6f6dccafd44fd9326302a4aaeecfacb1d8"
+checksum = "4219d74c6b67a3654a9fbebc4b419e22126d13d2f3c4a07ee0cb61ff79a79619"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -456,9 +456,9 @@ dependencies = [
 
 [[package]]
 name = "regex-automata"
-version = "0.4.8"
+version = "0.4.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "368758f23274712b504848e9d5a6f010445cc8b87a7cdb4d7cbee666c1288da3"
+checksum = "86b83b8b9847f9bf95ef68afb0b8e6cdb80f498442f5179a29fad448fcc1eaea"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -467,9 +467,9 @@ dependencies = [
 
 [[package]]
 name = "regex-syntax"
-version = "0.8.5"
+version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c"
+checksum = "adad44e29e4c806119491a7f06f03de4d1af22c3a680dd47f1e6e179439d1f56"
 
 [[package]]
 name = "ryu"

changelog.d/17749.doc

@@ -0,0 +1 @@
+Remove spurious "TODO UPDATE ALL THIS" note in the Debian installation docs.

debian/changelog

@@ -1,21 +1,3 @@
-matrix-synapse-py3 (1.117.0) stable; urgency=medium
-
-  * New Synapse release 1.117.0.
-
- -- Synapse Packaging team <packages@matrix.org>  Tue, 15 Oct 2024 10:46:30 +0100
-
-matrix-synapse-py3 (1.117.0~rc1) stable; urgency=medium
-
-  * New Synapse release 1.117.0rc1.
-
- -- Synapse Packaging team <packages@matrix.org>  Tue, 08 Oct 2024 14:37:11 +0100
-
-matrix-synapse-py3 (1.116.0) stable; urgency=medium
-
-  * New Synapse release 1.116.0.
-
- -- Synapse Packaging team <packages@matrix.org>  Tue, 01 Oct 2024 11:14:07 +0100
-
 matrix-synapse-py3 (1.116.0~rc2) stable; urgency=medium
 
   * New synapse release 1.116.0rc2.

docker/configure_workers_and_start.py

@@ -24,6 +24,15 @@
 # nginx and supervisord configs depending on the workers requested.
 #
 # The environment variables it reads are:
+#   * SYNAPSE_CONFIG_PATH: The path where the generated `homeserver.yaml` will
+#         be stored.
+#   * SYNAPSE_CONFIG_DIR: The directory where generated config will be stored.
+#         If `SYNAPSE_CONFIG_PATH` is not set, it will default to
+#         SYNAPSE_CONFIG_DIR/homeserver.yaml.
+#   * SYNAPSE_DATA_DIR: Where the generated config will put persistent data
+#         such as the database and media store.
+#   * SYNAPSE_CONFIG_TEMPLATE_DIR: The directory containing jinja2 templates for
+#         configuration that this script will generate config from. Defaults to '/conf'.
 #   * SYNAPSE_SERVER_NAME: The desired server_name of the homeserver.
 #   * SYNAPSE_REPORT_STATS: Whether to report stats.
 #   * SYNAPSE_WORKER_TYPES: A comma separated list of worker names as specified in WORKERS_CONFIG
@@ -35,6 +44,8 @@
 #         SYNAPSE_WORKER_TYPES='event_persister, federation_sender, client_reader'
 #         SYNAPSE_WORKER_TYPES='event_persister:2, federation_sender:2, client_reader'
 #         SYNAPSE_WORKER_TYPES='stream_writers=account_data+presence+typing'
+#   * SYNAPSE_WORKERS_WRITE_LOGS_TO_DISK: Whether worker logs should be written to disk,
+#         in addition to stdout.
 #   * SYNAPSE_AS_REGISTRATION_DIR: If specified, a directory in which .yaml and .yml files
 #         will be treated as Application Service registration files.
 #   * SYNAPSE_TLS_CERT: Path to a TLS certificate in PEM format.
@@ -48,7 +59,9 @@
 #   * SYNAPSE_LOG_SENSITIVE: If unset, SQL and SQL values won't be logged,
 #         regardless of the SYNAPSE_LOG_LEVEL setting.
 #   * SYNAPSE_LOG_TESTING: if set, Synapse will log additional information useful
-#     for testing.
+#         for testing.
+#   * SYNAPSE_USE_UNIX_SOCKET: if set, workers will communicate via unix socket
+#         rather than TCP.
 #
 # NOTE: According to Complement's ENTRYPOINT expectations for a homeserver image (as defined
 # in the project's README), this script may be run multiple times, and functionality should
@@ -604,7 +617,9 @@ def generate_base_homeserver_config() -> None:
     # start.py already does this for us, so just call that.
     # note that this script is copied in in the official, monolith dockerfile
     os.environ["SYNAPSE_HTTP_PORT"] = str(MAIN_PROCESS_HTTP_LISTENER_PORT)
-    subprocess.run(["/usr/local/bin/python", "/start.py", "migrate_config"], check=True)
+
+    # This script makes use of the `SYNAPSE_CONFIG_DIR` environment variable to
+    # determine where to place the generated homeserver config.
 
 
 def parse_worker_types(
@@ -733,8 +748,10 @@ def parse_worker_types(
 
 def generate_worker_files(
     environ: Mapping[str, str],
+    config_dir: str,
     config_path: str,
     data_dir: str,
+    template_dir: str,
     requested_worker_types: Dict[str, Set[str]],
 ) -> None:
     """Read the desired workers(if any) that is passed in and generate shared
@@ -742,9 +759,13 @@ def generate_worker_files(
 
     Args:
         environ: os.environ instance.
-        config_path: The location of the generated Synapse main worker config file.
-        data_dir: The location of the synapse data directory. Where log and
-            user-facing config files live.
+        config_dir: The location of the configuration directory, where generated
+            worker config files are written to.
+        config_path: The location of the base Synapse homeserver config file.
+        data_dir: The location of the synapse data directory. Where logs will be
+            stored (if `SYNAPSE_WORKERS_WRITE_LOGS_TO_DISK` is set).
+        template_dir: The location of the template directory. Where jinja2
+            templates for config files live.
         requested_worker_types: A Dict containing requested workers in the format of
             {'worker_name1': {'worker_type', ...}}
     """
@@ -807,7 +828,8 @@ def generate_worker_files(
     nginx_locations: Dict[str, str] = {}
 
     # Create the worker configuration directory if it doesn't already exist
-    os.makedirs("/conf/workers", exist_ok=True)
+    workers_config_dir = os.path.join(config_dir, "workers")
+    os.makedirs(workers_config_dir, exist_ok=True)
 
     # Start worker ports from this arbitrary port
     worker_port = 18009
@@ -854,7 +876,7 @@ def generate_worker_files(
         worker_config = insert_worker_name_for_worker_config(worker_config, worker_name)
 
         worker_config.update(
-            {"name": worker_name, "port": str(worker_port), "config_path": config_path}
+            {"name": worker_name, "port": str(worker_port)}
         )
 
         # Update the shared config with any worker_type specific options. The first of a
@@ -877,12 +899,14 @@ def generate_worker_files(
         worker_descriptors.append(worker_config)
 
         # Write out the worker's logging config file
-        log_config_filepath = generate_worker_log_config(environ, worker_name, data_dir)
+        log_config_filepath = generate_worker_log_config(
+            environ, worker_name, template_dir, workers_config_dir, data_dir
+        )
 
         # Then a worker config file
         convert(
-            "/conf/worker.yaml.j2",
-            f"/conf/workers/{worker_name}.yaml",
+            os.path.join(template_dir, "worker.yaml.j2"),
+            os.path.join(workers_config_dir, f"{worker_name}.yaml"),
             **worker_config,
             worker_log_config_filepath=log_config_filepath,
             using_unix_sockets=using_unix_sockets,
@@ -923,7 +947,9 @@ def generate_worker_files(
     # Finally, we'll write out the config files.
 
     # log config for the master process
-    master_log_config = generate_worker_log_config(environ, "master", data_dir)
+    master_log_config = generate_worker_log_config(
+        environ, "master", template_dir, workers_config_dir, data_dir
+    )
     shared_config["log_config"] = master_log_config
 
     # Find application service registrations
@@ -954,8 +980,8 @@ def generate_worker_files(
 
     # Shared homeserver config
     convert(
-        "/conf/shared.yaml.j2",
-        "/conf/workers/shared.yaml",
+        os.path.join(template_dir, "shared.yaml.j2"),
+        os.path.join(workers_config_dir, "shared.yaml"),
         shared_worker_config=yaml.dump(shared_config),
         appservice_registrations=appservice_registrations,
         enable_redis=workers_in_use,
@@ -965,7 +991,7 @@ def generate_worker_files(
 
     # Nginx config
     convert(
-        "/conf/nginx.conf.j2",
+        os.path.join(template_dir, "nginx.conf.j2"),
         "/etc/nginx/conf.d/matrix-synapse.conf",
         worker_locations=nginx_location_config,
         upstream_directives=nginx_upstream_config,
@@ -977,7 +1003,7 @@ def generate_worker_files(
     # Supervisord config
     os.makedirs("/etc/supervisor", exist_ok=True)
     convert(
-        "/conf/supervisord.conf.j2",
+        os.path.join(template_dir, "supervisord.conf.j2"),
         "/etc/supervisor/supervisord.conf",
         main_config_path=config_path,
         enable_redis=workers_in_use,
@@ -985,7 +1011,7 @@ def generate_worker_files(
     )
 
     convert(
-        "/conf/synapse.supervisord.conf.j2",
+        os.path.join(template_dir, "synapse.supervisord.conf.j2"),
         "/etc/supervisor/conf.d/synapse.conf",
         workers=worker_descriptors,
         main_config_path=config_path,
@@ -994,7 +1020,7 @@ def generate_worker_files(
 
     # healthcheck config
     convert(
-        "/conf/healthcheck.sh.j2",
+        os.path.join(template_dir, "healthcheck.sh.j2"),
         "/healthcheck.sh",
         healthcheck_urls=healthcheck_urls,
     )
@@ -1006,10 +1032,24 @@ def generate_worker_files(
 
 
 def generate_worker_log_config(
-    environ: Mapping[str, str], worker_name: str, data_dir: str
+    environ: Mapping[str, str],
+    worker_name: str,
+    workers_config_dir: str,
+    template_dir: str,
+    data_dir: str,
 ) -> str:
     """Generate a log.config file for the given worker.
 
+    Args:
+        environ: A mapping representing the environment variables that this script
+            is running with.
+        worker_name: The name of the worker. Used in generated file paths.
+        workers_config_dir: The location of the worker configuration directory,
+            where the generated worker log config will be saved.
+        template_dir: The directory containing jinja2 template files.
+        data_dir: The directory where log files will be written (if
+            `SYNAPSE_WORKERS_WRITE_LOGS_TO_DISK` is set).
+
     Returns: the path to the generated file
     """
     # Check whether we should write worker logs to disk, in addition to the console
@@ -1024,9 +1064,9 @@ def generate_worker_log_config(
     extra_log_template_args["SYNAPSE_LOG_TESTING"] = environ.get("SYNAPSE_LOG_TESTING")
 
     # Render and write the file
-    log_config_filepath = f"/conf/workers/{worker_name}.log.config"
+    log_config_filepath = os.path.join(workers_config_dir, f"{worker_name}.log.config")
     convert(
-        "/conf/log.config",
+        os.path.join(template_dir, "log.config"),
         log_config_filepath,
         worker_name=worker_name,
         **extra_log_template_args,
@@ -1049,6 +1089,7 @@ def main(args: List[str], environ: MutableMapping[str, str]) -> None:
     config_dir = environ.get("SYNAPSE_CONFIG_DIR", "/data")
     config_path = environ.get("SYNAPSE_CONFIG_PATH", config_dir + "/homeserver.yaml")
     data_dir = environ.get("SYNAPSE_DATA_DIR", "/data")
+    template_dir = environ.get("SYNAPSE_CONFIG_TEMPLATE_DIR", "/conf")
 
     # override SYNAPSE_NO_TLS, we don't support TLS in worker mode,
     # this needs to be handled by a frontend proxy
@@ -1060,9 +1101,10 @@ def main(args: List[str], environ: MutableMapping[str, str]) -> None:
         generate_base_homeserver_config()
     else:
         log("Base homeserver config exists—not regenerating")
+
     # This script may be run multiple times (mostly by Complement, see note at top of
     # file). Don't re-configure workers in this instance.
-    mark_filepath = "/conf/workers_have_been_configured"
+    mark_filepath = os.path.join(config_dir, "workers_have_been_configured")
     if not os.path.exists(mark_filepath):
         # Collect and validate worker_type requests
         # Read the desired worker configuration from the environment
@@ -1079,7 +1121,9 @@ def main(args: List[str], environ: MutableMapping[str, str]) -> None:
 
         # Always regenerate all other config files
         log("Generating worker config files")
-        generate_worker_files(environ, config_path, data_dir, requested_worker_types)
+        generate_worker_files(
+            environ, config_dir, config_path, data_dir, template_dir, requested_worker_types
+        )
 
         # Mark workers as being configured
         with open(mark_filepath, "w") as f:

docker/start.py

@@ -42,6 +42,8 @@ def convert(src: str, dst: str, environ: Mapping[str, object]) -> None:
 
 
 def generate_config_from_template(
+    data_dir: str,
+    template_dir: str,
     config_dir: str,
     config_path: str,
     os_environ: Mapping[str, str],
@@ -50,6 +52,9 @@ def generate_config_from_template(
     """Generate a homeserver.yaml from environment variables
 
     Args:
+        data_dir: where persistent data is stored
+        template_dir: The location of the template directory. Where jinja2
+            templates for config files live.
         config_dir: where to put generated config files
         config_path: where to put the main config file
         os_environ: environment mapping
@@ -70,9 +75,10 @@ def generate_config_from_template(
         "macaroon": "SYNAPSE_MACAROON_SECRET_KEY",
     }
 
+    synapse_server_name = environ["SYNAPSE_SERVER_NAME"]
     for name, secret in secrets.items():
         if secret not in environ:
-            filename = "/data/%s.%s.key" % (environ["SYNAPSE_SERVER_NAME"], name)
+            filename = os.path.join(data_dir, f"{synapse_server_name}.{name}.key")
 
             # if the file already exists, load in the existing value; otherwise,
             # generate a new secret and write it to a file
@@ -88,7 +94,7 @@ def generate_config_from_template(
                     handle.write(value)
             environ[secret] = value
 
-    environ["SYNAPSE_APPSERVICES"] = glob.glob("/data/appservices/*.yaml")
+    environ["SYNAPSE_APPSERVICES"] = glob.glob(os.path.join(data_dir, "appservices", "*.yaml"))
     if not os.path.exists(config_dir):
         os.mkdir(config_dir)
 
@@ -111,12 +117,12 @@ def generate_config_from_template(
         environ["SYNAPSE_LOG_CONFIG"] = config_dir + "/log.config"
 
     log("Generating synapse config file " + config_path)
-    convert("/conf/homeserver.yaml", config_path, environ)
+    convert(os.path.join(template_dir, "homeserver.yaml"), config_path, environ)
 
     log_config_file = environ["SYNAPSE_LOG_CONFIG"]
     log("Generating log config file " + log_config_file)
     convert(
-        "/conf/log.config",
+        os.path.join(template_dir, "log.config"),
         log_config_file,
         {**environ, "include_worker_name_in_log_line": False},
     )
@@ -128,15 +134,15 @@ def generate_config_from_template(
         "synapse.app.homeserver",
         "--config-path",
         config_path,
-        # tell synapse to put generated keys in /data rather than /compiled
+        # tell synapse to put generated keys in the data directory rather than /compiled
         "--keys-directory",
         config_dir,
         "--generate-keys",
     ]
 
     if ownership is not None:
-        log(f"Setting ownership on /data to {ownership}")
-        subprocess.run(["chown", "-R", ownership, "/data"], check=True)
+        log(f"Setting ownership on the data dir to {ownership}")
+        subprocess.run(["chown", "-R", ownership, data_dir], check=True)
         args = ["gosu", ownership] + args
 
     subprocess.run(args, check=True)
@@ -159,12 +165,13 @@ def run_generate_config(environ: Mapping[str, str], ownership: Optional[str]) ->
     config_dir = environ.get("SYNAPSE_CONFIG_DIR", "/data")
     config_path = environ.get("SYNAPSE_CONFIG_PATH", config_dir + "/homeserver.yaml")
     data_dir = environ.get("SYNAPSE_DATA_DIR", "/data")
+    template_dir = environ.get("SYNAPSE_CONFIG_TEMPLATE_DIR", "/conf")
 
     # create a suitable log config from our template
     log_config_file = "%s/%s.log.config" % (config_dir, server_name)
     if not os.path.exists(log_config_file):
         log("Creating log config %s" % (log_config_file,))
-        convert("/conf/log.config", log_config_file, environ)
+        convert(os.path.join(template_dir, "log.config"), log_config_file, environ)
 
     # generate the main config file, and a signing key.
     args = [
@@ -216,12 +223,14 @@ def main(args: List[str], environ: MutableMapping[str, str]) -> None:
 
     if mode == "migrate_config":
         # generate a config based on environment vars.
+        data_dir = environ.get("SYNAPSE_DATA_DIR", "/data")
         config_dir = environ.get("SYNAPSE_CONFIG_DIR", "/data")
         config_path = environ.get(
             "SYNAPSE_CONFIG_PATH", config_dir + "/homeserver.yaml"
         )
+        template_dir = environ.get("SYNAPSE_CONFIG_TEMPLATE_DIR", "/conf")
         return generate_config_from_template(
-            config_dir, config_path, environ, ownership
+            data_dir, template_dir, config_dir, config_path, environ, ownership
         )
 
     if mode != "run":

docs/usage/administration/admin_faq.md

@@ -255,8 +255,6 @@ line to `/etc/default/matrix-synapse`:
 
     LD_PRELOAD=/usr/lib/x86_64-linux-gnu/libjemalloc.so.2
 
-*Note*: You may need to set `PYTHONMALLOC=malloc` to ensure that `jemalloc` can accurately calculate memory usage. By default, Python uses its internal small-object allocator, which may interfere with jemalloc's ability to track memory consumption correctly. This could prevent the [cache_autotuning](../configuration/config_documentation.md#caches-and-associated-values) feature from functioning as expected, as the Python allocator may not reach the memory threshold set by `max_cache_memory_usage`, thus not triggering the cache eviction process.
-
 This made a significant difference on Python 2.7 - it's unclear how
 much of an improvement it provides on Python 3.x.
 

docs/usage/configuration/config_documentation.md

@@ -4368,13 +4368,7 @@ It is possible to scale the processes that handle sending outbound federation re
 by running a [`generic_worker`](../../workers.md#synapseappgeneric_worker) and adding it's [`worker_name`](#worker_name) to
 a `federation_sender_instances` map. Doing so will remove handling of this function from
 the main process. Multiple workers can be added to this map, in which case the work is
-balanced across them. 
-
-The way that the load balancing works is any outbound federation request will be assigned 
-to a federation sender worker based on the hash of the destination server name. This
-means that all requests being sent to the same destination will be processed by the same
-worker instance. Multiple `federation_sender_instances` are useful if there is a federation
-with multiple servers.
+balanced across them.
 
 This configuration setting must be shared between all workers handling federation
 sending, and if changed all federation sender workers must be stopped at the same time
@@ -4524,9 +4518,6 @@ This setting has the following sub-options:
 * `path`: The full path to a local Unix socket file. **If this is used, `host` and
  `port` are ignored.** Defaults to `/tmp/redis.sock'
 * `password`: Optional password if configured on the Redis instance.
-* `password_path`: Alternative to `password`, reading the password from an
-   external file. The file should be a plain text file, containing only the
-   password. Synapse reads the password from the given file once at startup.
 * `dbid`: Optional redis dbid if needs to connect to specific redis logical db.
 * `use_tls`: Whether to use tls connection. Defaults to false.
 * `certificate_file`: Optional path to the certificate file
@@ -4540,16 +4531,13 @@ This setting has the following sub-options:
 
   _Changed in Synapse 1.85.0: Added path option to use a local Unix socket_
 
-  _Changed in Synapse 1.116.0: Added password\_path_
-
 Example configuration:
 ```yaml
 redis:
   enabled: true
   host: localhost
   port: 6379
-  password_path: <path_to_the_password_file>
-  # OR password: <secret_password>
+  password: <secret_password>
   dbid: <dbid>
   #use_tls: True
   #certificate_file: <path_to_the_certificate_file>

poetry.lock

@@ -1447,13 +1447,13 @@ dev = ["jinja2"]
 
 [[package]]
 name = "phonenumbers"
-version = "8.13.46"
+version = "8.13.45"
 description = "Python version of Google's common library for parsing, formatting, storing and validating international phone numbers."
 optional = false
 python-versions = "*"
 files = [
-    {file = "phonenumbers-8.13.46-py2.py3-none-any.whl", hash = "sha256:519422d407af066fdbf98e179ea2e214487060f26526d67871f817eefbbb2134"},
-    {file = "phonenumbers-8.13.46.tar.gz", hash = "sha256:94bf18ba9725bb6868d29473b13f78ef01e2585c5cb561ec0200be7676e77452"},
+    {file = "phonenumbers-8.13.45-py2.py3-none-any.whl", hash = "sha256:bf05ec20fcd13f0d53e43a34ed7bd1c8be26a72b88fce4b8c64fca5b4641987a"},
+    {file = "phonenumbers-8.13.45.tar.gz", hash = "sha256:53679a95b6060fd5e15467759252c87933d8566d6a5be00995a579eb0e02435b"},
 ]
 
 [[package]]
@@ -1974,13 +1974,13 @@ six = ">=1.5"
 
 [[package]]
 name = "python-multipart"
-version = "0.0.12"
+version = "0.0.10"
 description = "A streaming multipart parser for Python"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "python_multipart-0.0.12-py3-none-any.whl", hash = "sha256:43dcf96cf65888a9cd3423544dd0d75ac10f7aa0c3c28a175bbcd00c9ce1aebf"},
-    {file = "python_multipart-0.0.12.tar.gz", hash = "sha256:045e1f98d719c1ce085ed7f7e1ef9d8ccc8c02ba02b5566d5f7521410ced58cb"},
+    {file = "python_multipart-0.0.10-py3-none-any.whl", hash = "sha256:2b06ad9e8d50c7a8db80e3b56dab590137b323410605af2be20d62a5f1ba1dc8"},
+    {file = "python_multipart-0.0.10.tar.gz", hash = "sha256:46eb3c6ce6fdda5fb1a03c7e11d490e407c6930a2703fe7aef4da71c374688fa"},
 ]
 
 [[package]]
@@ -2277,29 +2277,29 @@ files = [
 
 [[package]]
 name = "ruff"
-version = "0.6.8"
+version = "0.6.7"
 description = "An extremely fast Python linter and code formatter, written in Rust."
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "ruff-0.6.8-py3-none-linux_armv6l.whl", hash = "sha256:77944bca110ff0a43b768f05a529fecd0706aac7bcce36d7f1eeb4cbfca5f0f2"},
-    {file = "ruff-0.6.8-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:27b87e1801e786cd6ede4ada3faa5e254ce774de835e6723fd94551464c56b8c"},
-    {file = "ruff-0.6.8-py3-none-macosx_11_0_arm64.whl", hash = "sha256:cd48f945da2a6334f1793d7f701725a76ba93bf3d73c36f6b21fb04d5338dcf5"},
-    {file = "ruff-0.6.8-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:677e03c00f37c66cea033274295a983c7c546edea5043d0c798833adf4cf4c6f"},
-    {file = "ruff-0.6.8-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:9f1476236b3eacfacfc0f66aa9e6cd39f2a624cb73ea99189556015f27c0bdeb"},
-    {file = "ruff-0.6.8-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:6f5a2f17c7d32991169195d52a04c95b256378bbf0de8cb98478351eb70d526f"},
-    {file = "ruff-0.6.8-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:5fd0d4b7b1457c49e435ee1e437900ced9b35cb8dc5178921dfb7d98d65a08d0"},
-    {file = "ruff-0.6.8-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:f8034b19b993e9601f2ddf2c517451e17a6ab5cdb1c13fdff50c1442a7171d87"},
-    {file = "ruff-0.6.8-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:6cfb227b932ba8ef6e56c9f875d987973cd5e35bc5d05f5abf045af78ad8e098"},
-    {file = "ruff-0.6.8-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6ef0411eccfc3909269fed47c61ffebdcb84a04504bafa6b6df9b85c27e813b0"},
-    {file = "ruff-0.6.8-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:007dee844738c3d2e6c24ab5bc7d43c99ba3e1943bd2d95d598582e9c1b27750"},
-    {file = "ruff-0.6.8-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:ce60058d3cdd8490e5e5471ef086b3f1e90ab872b548814e35930e21d848c9ce"},
-    {file = "ruff-0.6.8-py3-none-musllinux_1_2_i686.whl", hash = "sha256:1085c455d1b3fdb8021ad534379c60353b81ba079712bce7a900e834859182fa"},
-    {file = "ruff-0.6.8-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:70edf6a93b19481affd287d696d9e311388d808671bc209fb8907b46a8c3af44"},
-    {file = "ruff-0.6.8-py3-none-win32.whl", hash = "sha256:792213f7be25316f9b46b854df80a77e0da87ec66691e8f012f887b4a671ab5a"},
-    {file = "ruff-0.6.8-py3-none-win_amd64.whl", hash = "sha256:ec0517dc0f37cad14a5319ba7bba6e7e339d03fbf967a6d69b0907d61be7a263"},
-    {file = "ruff-0.6.8-py3-none-win_arm64.whl", hash = "sha256:8d3bb2e3fbb9875172119021a13eed38849e762499e3cfde9588e4b4d70968dc"},
-    {file = "ruff-0.6.8.tar.gz", hash = "sha256:a5bf44b1aa0adaf6d9d20f86162b34f7c593bfedabc51239953e446aefc8ce18"},
+    {file = "ruff-0.6.7-py3-none-linux_armv6l.whl", hash = "sha256:08277b217534bfdcc2e1377f7f933e1c7957453e8a79764d004e44c40db923f2"},
+    {file = "ruff-0.6.7-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:c6707a32e03b791f4448dc0dce24b636cbcdee4dd5607adc24e5ee73fd86c00a"},
+    {file = "ruff-0.6.7-py3-none-macosx_11_0_arm64.whl", hash = "sha256:533d66b7774ef224e7cf91506a7dafcc9e8ec7c059263ec46629e54e7b1f90ab"},
+    {file = "ruff-0.6.7-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:17a86aac6f915932d259f7bec79173e356165518859f94649d8c50b81ff087e9"},
+    {file = "ruff-0.6.7-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:b3f8822defd260ae2460ea3832b24d37d203c3577f48b055590a426a722d50ef"},
+    {file = "ruff-0.6.7-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:9ba4efe5c6dbbb58be58dd83feedb83b5e95c00091bf09987b4baf510fee5c99"},
+    {file = "ruff-0.6.7-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:525201b77f94d2b54868f0cbe5edc018e64c22563da6c5c2e5c107a4e85c1c0d"},
+    {file = "ruff-0.6.7-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:8854450839f339e1049fdbe15d875384242b8e85d5c6947bb2faad33c651020b"},
+    {file = "ruff-0.6.7-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:2f0b62056246234d59cbf2ea66e84812dc9ec4540518e37553513392c171cb18"},
+    {file = "ruff-0.6.7-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6b1462fa56c832dc0cea5b4041cfc9c97813505d11cce74ebc6d1aae068de36b"},
+    {file = "ruff-0.6.7-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:02b083770e4cdb1495ed313f5694c62808e71764ec6ee5db84eedd82fd32d8f5"},
+    {file = "ruff-0.6.7-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:0c05fd37013de36dfa883a3854fae57b3113aaa8abf5dea79202675991d48624"},
+    {file = "ruff-0.6.7-py3-none-musllinux_1_2_i686.whl", hash = "sha256:f49c9caa28d9bbfac4a637ae10327b3db00f47d038f3fbb2195c4d682e925b14"},
+    {file = "ruff-0.6.7-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:a0e1655868164e114ba43a908fd2d64a271a23660195017c17691fb6355d59bb"},
+    {file = "ruff-0.6.7-py3-none-win32.whl", hash = "sha256:a939ca435b49f6966a7dd64b765c9df16f1faed0ca3b6f16acdf7731969deb35"},
+    {file = "ruff-0.6.7-py3-none-win_amd64.whl", hash = "sha256:590445eec5653f36248584579c06252ad2e110a5d1f32db5420de35fb0e1c977"},
+    {file = "ruff-0.6.7-py3-none-win_arm64.whl", hash = "sha256:b28f0d5e2f771c1fe3c7a45d3f53916fc74a480698c4b5731f0bea61e52137c8"},
+    {file = "ruff-0.6.7.tar.gz", hash = "sha256:44e52129d82266fa59b587e2cd74def5637b730a69c4542525dfdecfaae38bd5"},
 ]
 
 [[package]]
@@ -3114,4 +3114,4 @@ user-search = ["pyicu"]
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.8.0"
-content-hash = "304d03b74d2886def69ae44ce5afaed21318db9f09aae91281e0f182e1660ffd"
+content-hash = "93c267fac3428b764f954e6faa17937b9c97b1ed2bdafc41dd8f6cb5d2ce085b"

pyproject.toml

@@ -97,7 +97,7 @@ module-name = "synapse.synapse_rust"
 
 [tool.poetry]
 name = "matrix-synapse"
-version = "1.117.0"
+version = "1.116.0rc2"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "AGPL-3.0-or-later"
@@ -320,7 +320,7 @@ all = [
 # failing on new releases. Keeping lower bounds loose here means that dependabot
 # can bump versions without having to update the content-hash in the lockfile.
 # This helps prevents merge conflicts when running a batch of dependabot updates.
-ruff = "0.6.8"
+ruff = "0.6.7"
 # Type checking only works with the pydantic.v1 compat module from pydantic v2
 pydantic = "^2"
 

synapse/api/auth/msc3861_delegated.py

@@ -338,7 +338,7 @@ class MSC3861DelegatedAuth(BaseAuth):
             logger.exception("Failed to introspect token")
             raise SynapseError(503, "Unable to introspect the access token")
 
-        logger.debug("Introspection result: %r", introspection_result)
+        logger.info(f"Introspection result: {introspection_result!r}")
 
         # TODO: introspection verification should be more extensive, especially:
         #   - verify the audience

synapse/app/generic_worker.py

@@ -3,7 +3,7 @@
 #
 # Copyright 2020 The Matrix.org Foundation C.I.C.
 # Copyright 2016 OpenMarket Ltd
-# Copyright (C) 2023-2024 New Vector, Ltd
+# Copyright (C) 2023 New Vector, Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as

synapse/config/redis.py

@@ -21,15 +21,10 @@
 
 from typing import Any
 
-from synapse.config._base import Config, ConfigError, read_file
+from synapse.config._base import Config
 from synapse.types import JsonDict
 from synapse.util.check_dependencies import check_requirements
 
-CONFLICTING_PASSWORD_OPTS_ERROR = """\
-You have configured both `redis.password` and `redis.password_path`.
-These are mutually incompatible.
-"""
-
 
 class RedisConfig(Config):
     section = "redis"
@@ -48,17 +43,6 @@ class RedisConfig(Config):
         self.redis_path = redis_config.get("path", None)
         self.redis_dbid = redis_config.get("dbid", None)
         self.redis_password = redis_config.get("password")
-        redis_password_path = redis_config.get("password_path")
-        if redis_password_path:
-            if self.redis_password:
-                raise ConfigError(CONFLICTING_PASSWORD_OPTS_ERROR)
-            self.redis_password = read_file(
-                redis_password_path,
-                (
-                    "redis",
-                    "password_path",
-                ),
-            ).strip()
 
         self.redis_use_tls = redis_config.get("use_tls", False)
         self.redis_certificate = redis_config.get("certificate_file", None)

synapse/handlers/admin.py

@@ -443,8 +443,8 @@ class AdminHandler:
                 ["m.room.member", "m.room.message"],
             )
             if not event_ids:
-                # nothing to redact in this room
-                continue
+                # there's nothing to redact
+                return TaskStatus.COMPLETE, result, None
 
             events = await self._store.get_events_as_list(event_ids)
             for event in events:

synapse/handlers/delayed_events.py

@@ -1,17 +1,3 @@
-#
-# This file is licensed under the Affero General Public License (AGPL) version 3.
-#
-# Copyright (C) 2024 New Vector, Ltd
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# See the GNU Affero General Public License for more details:
-# <https://www.gnu.org/licenses/agpl-3.0.html>.
-#
-
 import logging
 from typing import TYPE_CHECKING, List, Optional, Set, Tuple
 

synapse/handlers/device.py

@@ -48,7 +48,6 @@ from synapse.metrics.background_process_metrics import (
     wrap_as_background_process,
 )
 from synapse.storage.databases.main.client_ips import DeviceLastConnectionInfo
-from synapse.storage.databases.main.roommember import EventIdMembership
 from synapse.storage.databases.main.state_deltas import StateDelta
 from synapse.types import (
     DeviceListUpdates,
@@ -223,6 +222,7 @@ class DeviceWorkerHandler:
         return changed
 
     @trace
+    @measure_func("device.get_user_ids_changed")
     @cancellable
     async def get_user_ids_changed(
         self, user_id: str, from_token: StreamToken
@@ -290,11 +290,9 @@ class DeviceWorkerHandler:
                 memberships_to_fetch.add(delta.prev_event_id)
 
         # Fetch all the memberships for the membership events
-        event_id_to_memberships: Mapping[str, Optional[EventIdMembership]] = {}
-        if memberships_to_fetch:
-            event_id_to_memberships = await self.store.get_membership_from_event_ids(
-                memberships_to_fetch
-            )
+        event_id_to_memberships = await self.store.get_membership_from_event_ids(
+            memberships_to_fetch
+        )
 
         joined_invited_knocked = (
             Membership.JOIN,
@@ -351,6 +349,7 @@ class DeviceWorkerHandler:
 
         return device_list_updates
 
+    @measure_func("_generate_sync_entry_for_device_list")
     async def generate_sync_entry_for_device_list(
         self,
         user_id: str,

synapse/handlers/sliding_sync/__init__.py

@@ -49,7 +49,6 @@ from synapse.types import (
     Requester,
     SlidingSyncStreamToken,
     StateMap,
-    StrCollection,
     StreamKeyType,
     StreamToken,
 )
@@ -294,6 +293,7 @@ class SlidingSyncHandler:
             # to record rooms as having updates even if there might not actually
             # be anything new for the user (e.g. due to event filters, events
             # having happened after the user left, etc).
+            unsent_room_ids = []
             if from_token:
                 # The set of rooms that the client (may) care about, but aren't
                 # in any list range (or subscribed to).
@@ -305,24 +305,15 @@ class SlidingSyncHandler:
                 # TODO: Replace this with something faster. When we land the
                 # sliding sync tables that record the most recent event
                 # positions we can use that.
-                unsent_room_ids: StrCollection
-                if await self.store.have_finished_sliding_sync_background_jobs():
-                    unsent_room_ids = await (
-                        self.store.get_rooms_that_have_updates_since_sliding_sync_table(
-                            room_ids=missing_rooms,
-                            from_key=from_token.stream_token.room_key,
-                        )
+                missing_event_map_by_room = (
+                    await self.store.get_room_events_stream_for_rooms(
+                        room_ids=missing_rooms,
+                        from_key=to_token.room_key,
+                        to_key=from_token.stream_token.room_key,
+                        limit=1,
                     )
-                else:
-                    missing_event_map_by_room = (
-                        await self.store.get_room_events_stream_for_rooms(
-                            room_ids=missing_rooms,
-                            from_key=to_token.room_key,
-                            to_key=from_token.stream_token.room_key,
-                            limit=1,
-                        )
-                    )
-                    unsent_room_ids = list(missing_event_map_by_room)
+                )
+                unsent_room_ids = list(missing_event_map_by_room)
 
                 new_connection_state.rooms.record_unsent_rooms(
                     unsent_room_ids, from_token.stream_token.room_key
@@ -1057,42 +1048,22 @@ class SlidingSyncHandler:
                     )
                 )
 
-        # Figure out the last bump event in the room. If the bump stamp hasn't
-        # changed we omit it from the response.
-        bump_stamp = None
-
-        always_return_bump_stamp = (
-            # We use the membership event position for any non-join
-            room_membership_for_user_at_to_token.membership != Membership.JOIN
-            # We didn't fetch any timeline events but we should still check for
-            # a bump_stamp that might be somewhere
-            or limited is None
-            # There might be a bump event somewhere before the timeline events
-            # that we fetched, that we didn't previously send down
-            or limited is True
-            # Always give the client some frame of reference if this is the
-            # first time they are seeing the room down the connection
-            or initial
-        )
-
+        # Figure out the last bump event in the room
+        #
+        # By default, just choose the membership event position for any non-join membership
+        bump_stamp = room_membership_for_user_at_to_token.event_pos.stream
         # If we're joined to the room, we need to find the last bump event before the
         # `to_token`
         if room_membership_for_user_at_to_token.membership == Membership.JOIN:
-            # Try and get a bump stamp
+            # Try and get a bump stamp, if not we just fall back to the
+            # membership token.
             new_bump_stamp = await self._get_bump_stamp(
-                room_id,
-                to_token,
-                timeline_events,
-                check_outside_timeline=always_return_bump_stamp,
+                room_id, to_token, timeline_events
             )
             if new_bump_stamp is not None:
                 bump_stamp = new_bump_stamp
 
-        if bump_stamp is None and always_return_bump_stamp:
-            # By default, just choose the membership event position for any non-join membership
-            bump_stamp = room_membership_for_user_at_to_token.event_pos.stream
-
-        if bump_stamp is not None and bump_stamp < 0:
+        if bump_stamp < 0:
             # We never want to send down negative stream orderings, as you can't
             # sensibly compare positive and negative stream orderings (they have
             # different meanings).
@@ -1185,23 +1156,14 @@ class SlidingSyncHandler:
 
     @trace
     async def _get_bump_stamp(
-        self,
-        room_id: str,
-        to_token: StreamToken,
-        timeline: List[EventBase],
-        check_outside_timeline: bool,
+        self, room_id: str, to_token: StreamToken, timeline: List[EventBase]
     ) -> Optional[int]:
-        """Get a bump stamp for the room, if we have a bump event and it has
-        changed.
+        """Get a bump stamp for the room, if we have a bump event
 
         Args:
             room_id
             to_token: The upper bound of token to return
             timeline: The list of events we have fetched.
-            limited: If the timeline was limited.
-            check_outside_timeline: Whether we need to check for bump stamp for
-                events before the timeline if we didn't find a bump stamp in
-                the timeline events.
         """
 
         # First check the timeline events we're returning to see if one of
@@ -1221,11 +1183,6 @@ class SlidingSyncHandler:
                 if new_bump_stamp > 0:
                     return new_bump_stamp
 
-        if not check_outside_timeline:
-            # If we are not a limited sync, then we know the bump stamp can't
-            # have changed.
-            return None
-
         # We can quickly query for the latest bump event in the room using the
         # sliding sync tables.
         latest_room_bump_stamp = await self.store.get_latest_bump_stamp_for_room(

synapse/handlers/sliding_sync/extensions.py

@@ -572,8 +572,7 @@ class SlidingSyncExtensionHandler:
 
             # Now record which rooms are now up to data, and which rooms have
             # pending updates to send.
-            new_connection_state.account_data.record_sent_rooms(previously_rooms.keys())
-            new_connection_state.account_data.record_sent_rooms(initial_rooms)
+            new_connection_state.account_data.record_sent_rooms(relevant_room_ids)
             missing_updates = (
                 all_updates_since_the_from_token.keys() - relevant_room_ids
             )
@@ -764,10 +763,9 @@ class SlidingSyncExtensionHandler:
 
                 room_id_to_receipt_map[room_id] = {"type": type, "content": content}
 
-            # Update the per-connection state to track which rooms we have sent
-            # all the receipts for.
-            new_connection_state.receipts.record_sent_rooms(previously_rooms.keys())
-            new_connection_state.receipts.record_sent_rooms(initial_rooms)
+        # Now we update the per-connection state to track which receipts we have
+        # and haven't sent down.
+        new_connection_state.receipts.record_sent_rooms(relevant_room_ids)
 
         if from_token:
             # Now find the set of rooms that may have receipts that we're not sending

synapse/handlers/sliding_sync/room_lists.py

@@ -123,19 +123,6 @@ class SlidingSyncInterestedRooms:
     newly_left_rooms: AbstractSet[str]
     dm_room_ids: AbstractSet[str]
 
-    @staticmethod
-    def empty() -> "SlidingSyncInterestedRooms":
-        return SlidingSyncInterestedRooms(
-            lists={},
-            relevant_room_map={},
-            relevant_rooms_to_send_map={},
-            all_rooms=set(),
-            room_membership_for_user_map={},
-            newly_joined_rooms=set(),
-            newly_left_rooms=set(),
-            dm_room_ids=set(),
-        )
-
 
 def filter_membership_for_sync(
     *,
@@ -194,14 +181,6 @@ class SlidingSyncRoomLists:
         from_token: Optional[StreamToken],
     ) -> SlidingSyncInterestedRooms:
         """Fetch the set of rooms that match the request"""
-        has_lists = sync_config.lists is not None and len(sync_config.lists) > 0
-        has_room_subscriptions = (
-            sync_config.room_subscriptions is not None
-            and len(sync_config.room_subscriptions) > 0
-        )
-
-        if not has_lists and not has_room_subscriptions:
-            return SlidingSyncInterestedRooms.empty()
 
         if await self.store.have_finished_sliding_sync_background_jobs():
             return await self._compute_interested_rooms_new_tables(
@@ -500,16 +479,6 @@ class SlidingSyncRoomLists:
                 # depending on the `required_state` requested (see below).
                 partial_state_rooms = await self.store.get_partial_rooms()
 
-                # Fetch any rooms that we have not already fetched from the database.
-                subscription_sliding_sync_rooms = (
-                    await self.store.get_sliding_sync_room_for_user_batch(
-                        user_id,
-                        sync_config.room_subscriptions.keys()
-                        - room_membership_for_user_map.keys(),
-                    )
-                )
-                room_membership_for_user_map.update(subscription_sliding_sync_rooms)
-
                 for (
                     room_id,
                     room_subscription,
@@ -517,11 +486,17 @@ class SlidingSyncRoomLists:
                     # Check if we have a membership for the room, but didn't pull it out
                     # above. This could be e.g. a leave that we don't pull out by
                     # default.
-                    current_room_entry = room_membership_for_user_map.get(room_id)
+                    current_room_entry = (
+                        await self.store.get_sliding_sync_room_for_user(
+                            user_id, room_id
+                        )
+                    )
                     if not current_room_entry:
                         # TODO: Handle rooms the user isn't in.
                         continue
 
+                    room_membership_for_user_map[room_id] = current_room_entry
+
                     all_rooms.add(room_id)
 
                     # Take the superset of the `RoomSyncConfig` for each room.

synapse/http/client.py

@@ -1039,7 +1039,7 @@ class _MultipartParserProtocol(protocol.Protocol):
         self.deferred = deferred
         self.boundary = boundary
         self.max_length = max_length
-        self.parser: Optional[multipart.MultipartParser] = None
+        self.parser = None
         self.multipart_response = MultipartResponse()
         self.has_redirect = False
         self.in_json = False
@@ -1097,7 +1097,7 @@ class _MultipartParserProtocol(protocol.Protocol):
                         self.deferred.errback()
                     self.file_length += end - start
 
-            callbacks: "multipart.multipart.MultipartCallbacks" = {
+            callbacks = {
                 "on_header_field": on_header_field,
                 "on_header_value": on_header_value,
                 "on_part_data": on_part_data,
@@ -1113,7 +1113,7 @@ class _MultipartParserProtocol(protocol.Protocol):
             self.transport.abortConnection()
 
         try:
-            self.parser.write(incoming_data)
+            self.parser.write(incoming_data)  # type: ignore[attr-defined]
         except Exception as e:
             logger.warning(f"Exception writing to multipart parser: {e}")
             self.deferred.errback()

synapse/notifier.py

@@ -41,7 +41,6 @@ import attr
 from prometheus_client import Counter
 
 from twisted.internet import defer
-from twisted.internet.defer import Deferred
 
 from synapse.api.constants import EduTypes, EventTypes, HistoryVisibility, Membership
 from synapse.api.errors import AuthError
@@ -53,7 +52,6 @@ from synapse.logging.opentracing import log_kv, start_active_span
 from synapse.metrics import LaterGauge
 from synapse.streams.config import PaginationConfig
 from synapse.types import (
-    ISynapseReactor,
     JsonDict,
     MultiWriterStreamToken,
     PersistedEventPosition,
@@ -63,11 +61,8 @@ from synapse.types import (
     StreamToken,
     UserID,
 )
-from synapse.util.async_helpers import (
-    timeout_deferred,
-)
+from synapse.util.async_helpers import ObservableDeferred, timeout_deferred
 from synapse.util.metrics import Measure
-from synapse.util.stringutils import shortstr
 from synapse.visibility import filter_events_for_client
 
 if TYPE_CHECKING:
@@ -94,6 +89,18 @@ def count(func: Callable[[T], bool], it: Iterable[T]) -> int:
     return n
 
 
+class _NotificationListener:
+    """This represents a single client connection to the events stream.
+    The events stream handler will have yielded to the deferred, so to
+    notify the handler it is sufficient to resolve the deferred.
+    """
+
+    __slots__ = ["deferred"]
+
+    def __init__(self, deferred: "defer.Deferred"):
+        self.deferred = deferred
+
+
 class _NotifierUserStream:
     """This represents a user connected to the event stream.
     It tracks the most recent stream token for that user.
@@ -106,49 +113,59 @@ class _NotifierUserStream:
 
     def __init__(
         self,
-        reactor: ISynapseReactor,
         user_id: str,
         rooms: StrCollection,
         current_token: StreamToken,
         time_now_ms: int,
     ):
-        self.reactor = reactor
         self.user_id = user_id
         self.rooms = set(rooms)
+        self.current_token = current_token
 
         # The last token for which we should wake up any streams that have a
         # token that comes before it. This gets updated every time we get poked.
         # We start it at the current token since if we get any streams
         # that have a token from before we have no idea whether they should be
         # woken up or not, so lets just wake them up.
-        self.current_token = current_token
+        self.last_notified_token = current_token
         self.last_notified_ms = time_now_ms
 
-        # Set of listeners that we need to wake up when there has been a change.
-        self.listeners: Set[Deferred[StreamToken]] = set()
+        self.notify_deferred: ObservableDeferred[StreamToken] = ObservableDeferred(
+            defer.Deferred()
+        )
 
-    def update_and_fetch_deferreds(
+    def notify(
         self,
-        current_token: StreamToken,
+        stream_key: StreamKeyType,
+        stream_id: Union[int, RoomStreamToken, MultiWriterStreamToken],
         time_now_ms: int,
-    ) -> Collection["Deferred[StreamToken]"]:
-        """Update the stream for this user because of a new event from an
-        event source, and return the set of deferreds to wake up.
-
+    ) -> None:
+        """Notify any listeners for this user of a new event from an
+        event source.
         Args:
-            current_token: The new current token.
+            stream_key: The stream the event came from.
+            stream_id: The new id for the stream the event came from.
             time_now_ms: The current time in milliseconds.
-
-        Returns:
-            The set of deferreds that need to be called.
         """
-        self.current_token = current_token
+        self.current_token = self.current_token.copy_and_advance(stream_key, stream_id)
+        self.last_notified_token = self.current_token
         self.last_notified_ms = time_now_ms
+        notify_deferred = self.notify_deferred
 
-        listeners = self.listeners
-        self.listeners = set()
+        log_kv(
+            {
+                "notify": self.user_id,
+                "stream": stream_key,
+                "stream_id": stream_id,
+                "listeners": self.count_listeners(),
+            }
+        )
 
-        return listeners
+        users_woken_by_stream_counter.labels(stream_key).inc()
+
+        with PreserveLoggingContext():
+            self.notify_deferred = ObservableDeferred(defer.Deferred())
+            notify_deferred.callback(self.current_token)
 
     def remove(self, notifier: "Notifier") -> None:
         """Remove this listener from all the indexes in the Notifier
@@ -162,9 +179,9 @@ class _NotifierUserStream:
         notifier.user_to_user_stream.pop(self.user_id)
 
     def count_listeners(self) -> int:
-        return len(self.listeners)
+        return len(self.notify_deferred.observers())
 
-    def new_listener(self, token: StreamToken) -> "Deferred[StreamToken]":
+    def new_listener(self, token: StreamToken) -> _NotificationListener:
         """Returns a deferred that is resolved when there is a new token
         greater than the given token.
 
@@ -174,17 +191,10 @@ class _NotifierUserStream:
         """
         # Immediately wake up stream if something has already since happened
         # since their last token.
-        if token != self.current_token:
-            return defer.succeed(self.current_token)
-
-        # Create a new deferred and add it to the set of listeners. We add a
-        # cancel handler to remove it from the set again, to handle timeouts.
-        deferred: "Deferred[StreamToken]" = Deferred(
-            canceller=lambda d: self.listeners.discard(d)
-        )
-        self.listeners.add(deferred)
-
-        return deferred
+        if self.last_notified_token != token:
+            return _NotificationListener(defer.succeed(self.current_token))
+        else:
+            return _NotificationListener(self.notify_deferred.observe())
 
 
 @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -237,7 +247,6 @@ class Notifier:
         # List of callbacks to be notified when a lock is released
         self._lock_released_callback: List[Callable[[str, str, str], None]] = []
 
-        self.reactor = hs.get_reactor()
         self.clock = hs.get_clock()
         self.appservice_handler = hs.get_application_service_handler()
         self._pusher_pool = hs.get_pusherpool()
@@ -333,25 +342,14 @@ class Notifier:
         # Wake up all related user stream notifiers
         user_streams = self.room_to_user_streams.get(room_id, set())
         time_now_ms = self.clock.time_msec()
-        current_token = self.event_sources.get_current_token()
-
-        listeners: List["Deferred[StreamToken]"] = []
         for user_stream in user_streams:
             try:
-                listeners.extend(
-                    user_stream.update_and_fetch_deferreds(current_token, time_now_ms)
+                user_stream.notify(
+                    StreamKeyType.UN_PARTIAL_STATED_ROOMS, new_token, time_now_ms
                 )
             except Exception:
                 logger.exception("Failed to notify listener")
 
-        with PreserveLoggingContext():
-            for listener in listeners:
-                listener.callback(current_token)
-
-        users_woken_by_stream_counter.labels(StreamKeyType.UN_PARTIAL_STATED_ROOMS).inc(
-            len(user_streams)
-        )
-
         # Poke the replication so that other workers also see the write to
         # the un-partial-stated rooms stream.
         self.notify_replication()
@@ -521,16 +519,12 @@ class Notifier:
         rooms = rooms or []
 
         with Measure(self.clock, "on_new_event"):
-            user_streams: Set[_NotifierUserStream] = set()
+            user_streams = set()
 
             log_kv(
                 {
                     "waking_up_explicit_users": len(users),
                     "waking_up_explicit_rooms": len(rooms),
-                    "users": shortstr(users),
-                    "rooms": shortstr(rooms),
-                    "stream": stream_key,
-                    "stream_id": new_token,
                 }
             )
 
@@ -550,27 +544,12 @@ class Notifier:
                 )
 
             time_now_ms = self.clock.time_msec()
-            current_token = self.event_sources.get_current_token()
-            listeners: List["Deferred[StreamToken]"] = []
             for user_stream in user_streams:
                 try:
-                    listeners.extend(
-                        user_stream.update_and_fetch_deferreds(
-                            current_token, time_now_ms
-                        )
-                    )
+                    user_stream.notify(stream_key, new_token, time_now_ms)
                 except Exception:
                     logger.exception("Failed to notify listener")
 
-            # We resolve all these deferreds in one go so that we only need to
-            # call `PreserveLoggingContext` once, as it has a bunch of overhead
-            # (to calculate performance stats)
-            with PreserveLoggingContext():
-                for listener in listeners:
-                    listener.callback(current_token)
-
-            users_woken_by_stream_counter.labels(stream_key).inc(len(user_streams))
-
             self.notify_replication()
 
             # Notify appservices.
@@ -607,7 +586,6 @@ class Notifier:
             if room_ids is None:
                 room_ids = await self.store.get_rooms_for_user(user_id)
             user_stream = _NotifierUserStream(
-                reactor=self.reactor,
                 user_id=user_id,
                 rooms=room_ids,
                 current_token=current_token,
@@ -630,8 +608,8 @@ class Notifier:
                         # Now we wait for the _NotifierUserStream to be told there
                         # is a new token.
                         listener = user_stream.new_listener(prev_token)
-                        listener = timeout_deferred(
-                            listener,
+                        listener.deferred = timeout_deferred(
+                            listener.deferred,
                             (end_time - now) / 1000.0,
                             self.hs.get_reactor(),
                         )
@@ -644,7 +622,7 @@ class Notifier:
                         )
 
                         with PreserveLoggingContext():
-                            await listener
+                            await listener.deferred
 
                         log_kv(
                             {

synapse/replication/http/__init__.py

@@ -1,7 +1,7 @@
 #
 # This file is licensed under the Affero General Public License (AGPL) version 3.
 #
-# Copyright (C) 2023-2024 New Vector, Ltd
+# Copyright (C) 2023 New Vector, Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as

synapse/replication/http/delayed_events.py

@@ -1,17 +1,3 @@
-#
-# This file is licensed under the Affero General Public License (AGPL) version 3.
-#
-# Copyright (C) 2024 New Vector, Ltd
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# See the GNU Affero General Public License for more details:
-# <https://www.gnu.org/licenses/agpl-3.0.html>.
-#
-
 import logging
 from typing import TYPE_CHECKING, Dict, Optional, Tuple
 

synapse/replication/http/push.py

@@ -48,7 +48,7 @@ class ReplicationRemovePusherRestServlet(ReplicationEndpoint):
 
     """
 
-    NAME = "remove_pusher"
+    NAME = "add_user_account_data"
     PATH_ARGS = ("user_id",)
     CACHE = False
 

synapse/rest/admin/users.py

@@ -27,7 +27,7 @@ from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
 
 import attr
 
-from synapse._pydantic_compat import StrictBool, StrictInt, StrictStr
+from synapse._pydantic_compat import StrictBool
 from synapse.api.constants import Direction, UserTypes
 from synapse.api.errors import Codes, NotFoundError, SynapseError
 from synapse.http.servlet import (
@@ -1421,39 +1421,40 @@ class RedactUser(RestServlet):
         self._store = hs.get_datastores().main
         self.admin_handler = hs.get_admin_handler()
 
-    class PostBody(RequestBodyModel):
-        rooms: List[StrictStr]
-        reason: Optional[StrictStr]
-        limit: Optional[StrictInt]
-
     async def on_POST(
         self, request: SynapseRequest, user_id: str
     ) -> Tuple[int, JsonDict]:
         requester = await self._auth.get_user_by_req(request)
         await assert_user_is_admin(self._auth, requester)
 
-        # parse provided user id to check that it is valid
-        UserID.from_string(user_id)
-
-        body = parse_and_validate_json_object_from_request(request, self.PostBody)
-
-        limit = body.limit
-        if limit and limit <= 0:
+        body = parse_json_object_from_request(request, allow_empty_body=True)
+        rooms = body.get("rooms")
+        if rooms is None:
             raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "If limit is provided it must be a non-negative integer greater than 0.",
+                HTTPStatus.BAD_REQUEST, "Must provide a value for rooms."
             )
 
-        rooms = body.rooms
+        reason = body.get("reason")
+        if reason:
+            if not isinstance(reason, str):
+                raise SynapseError(
+                    HTTPStatus.BAD_REQUEST,
+                    "If a reason is provided it must be a string.",
+                )
+
+        limit = body.get("limit")
+        if limit:
+            if not isinstance(limit, int) or limit <= 0:
+                raise SynapseError(
+                    HTTPStatus.BAD_REQUEST,
+                    "If limit is provided it must be a non-negative integer greater than 0.",
+                )
+
         if not rooms:
-            current_rooms = list(await self._store.get_rooms_for_user(user_id))
-            banned_rooms = list(
-                await self._store.get_rooms_user_currently_banned_from(user_id)
-            )
-            rooms = current_rooms + banned_rooms
+            rooms = await self._store.get_rooms_for_user(user_id)
 
         redact_id = await self.admin_handler.start_redact_events(
-            user_id, rooms, requester.serialize(), body.reason, limit
+            user_id, list(rooms), requester.serialize(), reason, limit
         )
 
         return HTTPStatus.OK, {"redact_id": redact_id}

synapse/rest/client/delayed_events.py

@@ -1,17 +1,3 @@
-#
-# This file is licensed under the Affero General Public License (AGPL) version 3.
-#
-# Copyright (C) 2024 New Vector, Ltd
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# See the GNU Affero General Public License for more details:
-# <https://www.gnu.org/licenses/agpl-3.0.html>.
-#
-
 # This module contains REST servlets to do with delayed events: /delayed_events/<paths>
 
 import logging

synapse/rest/client/sync.py

@@ -1010,13 +1010,11 @@ class SlidingSyncRestServlet(RestServlet):
         serialized_rooms: Dict[str, JsonDict] = {}
         for room_id, room_result in rooms.items():
             serialized_rooms[room_id] = {
+                "bump_stamp": room_result.bump_stamp,
                 "notification_count": room_result.notification_count,
                 "highlight_count": room_result.highlight_count,
             }
 
-            if room_result.bump_stamp is not None:
-                serialized_rooms[room_id]["bump_stamp"] = room_result.bump_stamp
-
             if room_result.joined_count is not None:
                 serialized_rooms[room_id]["joined_count"] = room_result.joined_count
 

synapse/rest/client/versions.py

@@ -172,7 +172,7 @@ class VersionsRestServlet(RestServlet):
                         )
                     ),
                     # MSC4140: Delayed events
-                    "org.matrix.msc4140": bool(self.config.server.max_event_delay_ms),
+                    "org.matrix.msc4140": True,
                     # MSC4151: Report room API (Client-Server API)
                     "org.matrix.msc4151": self.config.experimental.msc4151_enabled,
                     # Simplified sliding sync

synapse/storage/databases/main/delayed_events.py

@@ -1,17 +1,3 @@
-#
-# This file is licensed under the Affero General Public License (AGPL) version 3.
-#
-# Copyright (C) 2024 New Vector, Ltd
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# See the GNU Affero General Public License for more details:
-# <https://www.gnu.org/licenses/agpl-3.0.html>.
-#
-
 import logging
 from typing import List, NewType, Optional, Tuple
 

synapse/storage/databases/main/roommember.py

@@ -711,27 +711,6 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
 
         return {row[0] for row in txn}
 
-    async def get_rooms_user_currently_banned_from(
-        self, user_id: str
-    ) -> FrozenSet[str]:
-        """Returns a set of room_ids the user is currently banned from.
-
-        If a remote user only returns rooms this server is currently
-        participating in.
-        """
-        room_ids = await self.db_pool.simple_select_onecol(
-            table="current_state_events",
-            keyvalues={
-                "type": EventTypes.Member,
-                "membership": Membership.BAN,
-                "state_key": user_id,
-            },
-            retcol="room_id",
-            desc="get_rooms_user_currently_banned_from",
-        )
-
-        return frozenset(room_ids)
-
     @cached(max_entries=500000, iterable=True)
     async def get_rooms_for_user(self, user_id: str) -> FrozenSet[str]:
         """Returns a set of room_ids the user is currently joined to.
@@ -1520,57 +1499,6 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
             "get_sliding_sync_room_for_user", get_sliding_sync_room_for_user_txn
         )
 
-    async def get_sliding_sync_room_for_user_batch(
-        self, user_id: str, room_ids: StrCollection
-    ) -> Dict[str, RoomsForUserSlidingSync]:
-        """Get the sliding sync room entry for the given user and rooms."""
-
-        if not room_ids:
-            return {}
-
-        def get_sliding_sync_room_for_user_batch_txn(
-            txn: LoggingTransaction,
-        ) -> Dict[str, RoomsForUserSlidingSync]:
-            clause, args = make_in_list_sql_clause(
-                self.database_engine, "m.room_id", room_ids
-            )
-            sql = f"""
-                SELECT m.room_id, m.sender, m.membership, m.membership_event_id,
-                    r.room_version,
-                    m.event_instance_name, m.event_stream_ordering,
-                    m.has_known_state,
-                    COALESCE(j.room_type, m.room_type),
-                    COALESCE(j.is_encrypted, m.is_encrypted)
-                FROM sliding_sync_membership_snapshots AS m
-                INNER JOIN rooms AS r USING (room_id)
-                LEFT JOIN sliding_sync_joined_rooms AS j ON (j.room_id = m.room_id AND m.membership = 'join')
-                WHERE m.forgotten = 0
-                    AND {clause}
-                    AND user_id = ?
-            """
-            args.append(user_id)
-            txn.execute(sql, args)
-
-            return {
-                row[0]: RoomsForUserSlidingSync(
-                    room_id=row[0],
-                    sender=row[1],
-                    membership=row[2],
-                    event_id=row[3],
-                    room_version_id=row[4],
-                    event_pos=PersistedEventPosition(row[5], row[6]),
-                    has_known_state=bool(row[7]),
-                    room_type=row[8],
-                    is_encrypted=row[9],
-                )
-                for row in txn
-            }
-
-        return await self.db_pool.runInteraction(
-            "get_sliding_sync_room_for_user_batch",
-            get_sliding_sync_room_for_user_batch_txn,
-        )
-
 
 class RoomMemberBackgroundUpdateStore(SQLBaseStore):
     def __init__(

synapse/storage/databases/main/stream.py

@@ -751,48 +751,6 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             if self._events_stream_cache.has_entity_changed(room_id, from_id)
         }
 
-    async def get_rooms_that_have_updates_since_sliding_sync_table(
-        self,
-        room_ids: StrCollection,
-        from_key: RoomStreamToken,
-    ) -> StrCollection:
-        """Return the rooms that probably have had updates since the given
-        token (changes that are > `from_key`)."""
-        # If the stream change cache is valid for the stream token, we can just
-        # use the result of that.
-        if from_key.stream >= self._events_stream_cache.get_earliest_known_position():
-            return self._events_stream_cache.get_entities_changed(
-                room_ids, from_key.stream
-            )
-
-        def get_rooms_that_have_updates_since_sliding_sync_table_txn(
-            txn: LoggingTransaction,
-        ) -> StrCollection:
-            sql = """
-                SELECT room_id
-                FROM sliding_sync_joined_rooms
-                WHERE {clause}
-                    AND event_stream_ordering > ?
-            """
-
-            results: Set[str] = set()
-            for batch in batch_iter(room_ids, 1000):
-                clause, args = make_in_list_sql_clause(
-                    self.database_engine, "room_id", batch
-                )
-
-                args.append(from_key.stream)
-                txn.execute(sql.format(clause=clause), args)
-
-                results.update(row[0] for row in txn)
-
-            return results
-
-        return await self.db_pool.runInteraction(
-            "get_rooms_that_have_updates_since_sliding_sync_table",
-            get_rooms_that_have_updates_since_sliding_sync_table_txn,
-        )
-
     async def paginate_room_events_by_stream_ordering(
         self,
         *,

synapse/storage/schema/main/delta/88/01_add_delayed_events.sql

@@ -1,16 +1,3 @@
---
--- This file is licensed under the Affero General Public License (AGPL) version 3.
---
--- Copyright (C) 2024 New Vector, Ltd
---
--- This program is free software: you can redistribute it and/or modify
--- it under the terms of the GNU Affero General Public License as
--- published by the Free Software Foundation, either version 3 of the
--- License, or (at your option) any later version.
---
--- See the GNU Affero General Public License for more details:
--- <https://www.gnu.org/licenses/agpl-3.0.html>.
-
 CREATE TABLE delayed_events (
     delay_id TEXT NOT NULL,
     user_localpart TEXT NOT NULL,

synapse/types/handlers/sliding_sync.py

@@ -158,7 +158,6 @@ class SlidingSyncResult:
                 name changes to mark the room as unread and bump it to the top. For
                 encrypted rooms, we just have to consider any activity as a bump because we
                 can't see the content and the client has to figure it out for themselves.
-                This may not be included if there hasn't been a change.
             joined_count: The number of users with membership of join, including the client's
                 own user ID. (same as sync `v2 m.joined_member_count`)
             invited_count: The number of users with membership of invite. (same as sync v2
@@ -194,7 +193,7 @@ class SlidingSyncResult:
         limited: Optional[bool]
         # Only optional because it won't be included for invite/knock rooms with `stripped_state`
         num_live: Optional[int]
-        bump_stamp: Optional[int]
+        bump_stamp: int
         joined_count: Optional[int]
         invited_count: Optional[int]
         notification_count: int

synapse/util/caches/stream_change_cache.py

@@ -142,9 +142,9 @@ class StreamChangeCache:
         """
         assert isinstance(stream_pos, int)
 
-        # _cache is not valid before the earliest known stream position, so
+        # _cache is not valid at or before the earliest known stream position, so
         # return that the entity has changed.
-        if stream_pos < self._earliest_known_stream_pos:
+        if stream_pos <= self._earliest_known_stream_pos:
             self.metrics.inc_misses()
             return True
 
@@ -186,7 +186,7 @@ class StreamChangeCache:
             This will be all entities if the given stream position is at or earlier
             than the earliest known stream position.
         """
-        if not self._cache or stream_pos < self._earliest_known_stream_pos:
+        if not self._cache or stream_pos <= self._earliest_known_stream_pos:
             self.metrics.inc_misses()
             return set(entities)
 
@@ -238,9 +238,9 @@ class StreamChangeCache:
         """
         assert isinstance(stream_pos, int)
 
-        # _cache is not valid before the earliest known stream position, so
+        # _cache is not valid at or before the earliest known stream position, so
         # return that an entity has changed.
-        if stream_pos < self._earliest_known_stream_pos:
+        if stream_pos <= self._earliest_known_stream_pos:
             self.metrics.inc_misses()
             return True
 
@@ -270,9 +270,9 @@ class StreamChangeCache:
         """
         assert isinstance(stream_pos, int)
 
-        # _cache is not valid before the earliest known stream position, so
+        # _cache is not valid at or before the earliest known stream position, so
         # return None to mark that it is unknown if an entity has changed.
-        if stream_pos < self._earliest_known_stream_pos:
+        if stream_pos <= self._earliest_known_stream_pos:
             return AllEntitiesChangedResult(None)
 
         changed_entities: List[EntityType] = []

tests/config/test_load.py

@@ -19,23 +19,13 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-import tempfile
-from typing import Callable
-
 import yaml
-from parameterized import parameterized
 
 from synapse.config import ConfigError
-from synapse.config._base import RootConfig
 from synapse.config.homeserver import HomeServerConfig
 
 from tests.config.utils import ConfigFileTestCase
 
-try:
-    import hiredis
-except ImportError:
-    hiredis = None  # type: ignore
-
 
 class ConfigLoadingFileTestCase(ConfigFileTestCase):
     def test_load_fails_if_server_name_missing(self) -> None:
@@ -126,49 +116,3 @@ class ConfigLoadingFileTestCase(ConfigFileTestCase):
         self.add_lines_to_config(["trust_identity_server_for_password_resets: true"])
         with self.assertRaises(ConfigError):
             HomeServerConfig.load_config("", ["-c", self.config_file])
-
-    @parameterized.expand(
-        [
-            "turn_shared_secret_path: /does/not/exist",
-            "registration_shared_secret_path: /does/not/exist",
-            *["redis:\n  enabled: true\n  password_path: /does/not/exist"]
-            * (hiredis is not None),
-        ]
-    )
-    def test_secret_files_missing(self, config_str: str) -> None:
-        self.generate_config()
-        self.add_lines_to_config(["", config_str])
-
-        with self.assertRaises(ConfigError):
-            HomeServerConfig.load_config("", ["-c", self.config_file])
-
-    @parameterized.expand(
-        [
-            (
-                "turn_shared_secret_path: {}",
-                lambda c: c.voip.turn_shared_secret,
-            ),
-            (
-                "registration_shared_secret_path: {}",
-                lambda c: c.registration.registration_shared_secret,
-            ),
-            *[
-                (
-                    "redis:\n  enabled: true\n  password_path: {}",
-                    lambda c: c.redis.redis_password,
-                )
-            ]
-            * (hiredis is not None),
-        ]
-    )
-    def test_secret_files_existing(
-        self, config_line: str, get_secret: Callable[[RootConfig], str]
-    ) -> None:
-        self.generate_config_and_remove_lines_containing("registration_shared_secret")
-        with tempfile.NamedTemporaryFile(buffering=0) as secret_file:
-            secret_file.write(b"53C237")
-
-            self.add_lines_to_config(["", config_line.format(secret_file.name)])
-            config = HomeServerConfig.load_config("", ["-c", self.config_file])
-
-            self.assertEqual(get_secret(config), "53C237")

tests/handlers/test_room_member.py

@@ -380,7 +380,7 @@ class RoomMemberMasterHandlerTestCase(HomeserverTestCase):
         )
 
     def test_forget_when_not_left(self) -> None:
-        """Tests that a user cannot forget a room that they are still in."""
+        """Tests that a user cannot not forgets a room that has not left."""
         self.get_failure(self.handler.forget(self.alice_ID, self.room_id), SynapseError)
 
     def test_nonlocal_room_user_action(self) -> None:

tests/rest/admin/test_user.py

@@ -5288,26 +5288,19 @@ class UserRedactionTestCase(unittest.HomeserverTestCase):
         self.assertEqual(len(matched), len(rm2_originals))
 
     def test_admin_redact_works_if_user_kicked_or_banned(self) -> None:
-        originals1 = []
-        originals2 = []
+        originals = []
         for rm in [self.rm1, self.rm2, self.rm3]:
             join = self.helper.join(rm, self.bad_user, tok=self.bad_user_tok)
-            if rm in [self.rm1, self.rm3]:
-                originals1.append(join["event_id"])
-            else:
-                originals2.append(join["event_id"])
+            originals.append(join["event_id"])
             for i in range(5):
                 event = {"body": f"hello{i}", "msgtype": "m.text"}
                 res = self.helper.send_event(
                     rm, "m.room.message", event, tok=self.bad_user_tok
                 )
-                if rm in [self.rm1, self.rm3]:
-                    originals1.append(res["event_id"])
-                else:
-                    originals2.append(res["event_id"])
+                originals.append(res["event_id"])
 
         # kick user from rooms 1 and 3
-        for r in [self.rm1, self.rm3]:
+        for r in [self.rm1, self.rm2]:
             channel = self.make_request(
                 "POST",
                 f"/_matrix/client/r0/rooms/{r}/kick",
@@ -5337,70 +5330,32 @@ class UserRedactionTestCase(unittest.HomeserverTestCase):
         failed_redactions = channel2.json_body.get("failed_redactions")
         self.assertEqual(failed_redactions, {})
 
-        # double check
-        for rm in [self.rm1, self.rm3]:
-            filter = json.dumps({"types": [EventTypes.Redaction]})
-            channel3 = self.make_request(
-                "GET",
-                f"rooms/{rm}/messages?filter={filter}&limit=50",
-                access_token=self.admin_tok,
-            )
-            self.assertEqual(channel3.code, 200)
-
-            matches = []
-            for event in channel3.json_body["chunk"]:
-                for event_id in originals1:
-                    if (
-                        event["type"] == "m.room.redaction"
-                        and event["redacts"] == event_id
-                    ):
-                        matches.append((event_id, event))
-            # we redacted 6 messages
-            self.assertEqual(len(matches), 6)
-
-        # ban user from room 2
-        channel4 = self.make_request(
+        # ban user
+        channel3 = self.make_request(
             "POST",
             f"/_matrix/client/r0/rooms/{self.rm2}/ban",
             content={"reason": "being a bummer", "user_id": self.bad_user},
             access_token=self.admin_tok,
         )
-        self.assertEqual(channel4.code, HTTPStatus.OK, channel4.result)
+        self.assertEqual(channel3.code, HTTPStatus.OK, channel3.result)
 
-        # make a request to ban all user's messages
-        channel5 = self.make_request(
+        # redact messages in room 2
+        channel4 = self.make_request(
             "POST",
             f"/_synapse/admin/v1/user/{self.bad_user}/redact",
-            content={"rooms": []},
+            content={"rooms": [self.rm2]},
             access_token=self.admin_tok,
         )
-        self.assertEqual(channel5.code, 200)
-        id2 = channel5.json_body.get("redact_id")
+        self.assertEqual(channel4.code, 200)
+        id2 = channel1.json_body.get("redact_id")
 
         # check that there were no failed redactions in room 2
-        channel6 = self.make_request(
+        channel5 = self.make_request(
             "GET",
             f"/_synapse/admin/v1/user/redact_status/{id2}",
             access_token=self.admin_tok,
         )
-        self.assertEqual(channel6.code, 200)
-        self.assertEqual(channel6.json_body.get("status"), "complete")
-        failed_redactions = channel6.json_body.get("failed_redactions")
+        self.assertEqual(channel5.code, 200)
+        self.assertEqual(channel5.json_body.get("status"), "complete")
+        failed_redactions = channel5.json_body.get("failed_redactions")
         self.assertEqual(failed_redactions, {})
-
-        # double check messages in room 2 were redacted
-        filter = json.dumps({"types": [EventTypes.Redaction]})
-        channel7 = self.make_request(
-            "GET",
-            f"rooms/{self.rm2}/messages?filter={filter}&limit=50",
-            access_token=self.admin_tok,
-        )
-        self.assertEqual(channel7.code, 200)
-
-        matches = []
-        for event in channel7.json_body["chunk"]:
-            for event_id in originals2:
-                if event["type"] == "m.room.redaction" and event["redacts"] == event_id:
-                    matches.append((event_id, event))
-        # we redacted 6 messages
-        self.assertEqual(len(matches), 6)

tests/rest/client/sliding_sync/test_rooms_meta.py

@@ -1096,92 +1096,6 @@ class SlidingSyncRoomsMetaTestCase(SlidingSyncBase):
 
         self.assertGreater(response_body["rooms"][room_id]["bump_stamp"], 0)
 
-    def test_rooms_bump_stamp_no_change_incremental(self) -> None:
-        """Test that the bump stamp is omitted if there has been no change"""
-
-        user1_id = self.register_user("user1", "pass")
-        user1_tok = self.login(user1_id, "pass")
-
-        room_id1 = self.helper.create_room_as(
-            user1_id,
-            tok=user1_tok,
-        )
-
-        # Make the Sliding Sync request
-        sync_body = {
-            "lists": {
-                "foo-list": {
-                    "ranges": [[0, 1]],
-                    "required_state": [],
-                    "timeline_limit": 100,
-                }
-            }
-        }
-        response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
-
-        # Initial sync so we expect to see a bump stamp
-        self.assertIn("bump_stamp", response_body["rooms"][room_id1])
-
-        # Send an event that is not in the bump events list
-        self.helper.send_event(
-            room_id1, type="org.matrix.test", content={}, tok=user1_tok
-        )
-
-        response_body, from_token = self.do_sync(
-            sync_body, since=from_token, tok=user1_tok
-        )
-
-        # There hasn't been a change to the bump stamps, so we ignore it
-        self.assertNotIn("bump_stamp", response_body["rooms"][room_id1])
-
-    def test_rooms_bump_stamp_change_incremental(self) -> None:
-        """Test that the bump stamp is included if there has been a change, even
-        if its not in the timeline"""
-
-        user1_id = self.register_user("user1", "pass")
-        user1_tok = self.login(user1_id, "pass")
-
-        room_id1 = self.helper.create_room_as(
-            user1_id,
-            tok=user1_tok,
-        )
-
-        # Make the Sliding Sync request
-        sync_body = {
-            "lists": {
-                "foo-list": {
-                    "ranges": [[0, 1]],
-                    "required_state": [],
-                    "timeline_limit": 2,
-                }
-            }
-        }
-        response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
-
-        # Initial sync so we expect to see a bump stamp
-        self.assertIn("bump_stamp", response_body["rooms"][room_id1])
-        first_bump_stamp = response_body["rooms"][room_id1]["bump_stamp"]
-
-        # Send a bump event at the start.
-        self.helper.send(room_id1, "test", tok=user1_tok)
-
-        # Send events that are not in the bump events list to fill the timeline
-        for _ in range(5):
-            self.helper.send_event(
-                room_id1, type="org.matrix.test", content={}, tok=user1_tok
-            )
-
-        response_body, from_token = self.do_sync(
-            sync_body, since=from_token, tok=user1_tok
-        )
-
-        # There was a bump event in the timeline gap, so we should see the bump
-        # stamp be updated.
-        self.assertIn("bump_stamp", response_body["rooms"][room_id1])
-        second_bump_stamp = response_body["rooms"][room_id1]["bump_stamp"]
-
-        self.assertGreater(second_bump_stamp, first_bump_stamp)
-
     def test_rooms_bump_stamp_invites(self) -> None:
         """
         Test that `bump_stamp` is present and points to the membership event,

tests/rest/client/test_delayed_events.py

@@ -1,17 +1,3 @@
-#
-# This file is licensed under the Affero General Public License (AGPL) version 3.
-#
-# Copyright (C) 2024 New Vector, Ltd
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as
-# published by the Free Software Foundation, either version 3 of the
-# License, or (at your option) any later version.
-#
-# See the GNU Affero General Public License for more details:
-# <https://www.gnu.org/licenses/agpl-3.0.html>.
-#
-
 """Tests REST events for /delayed_events paths."""
 
 from http import HTTPStatus
@@ -22,12 +8,11 @@ from parameterized import parameterized
 from twisted.test.proto_helpers import MemoryReactor
 
 from synapse.api.errors import Codes
-from synapse.rest.client import delayed_events, room, versions
+from synapse.rest.client import delayed_events, room
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util import Clock
 
-from tests import unittest
 from tests.unittest import HomeserverTestCase
 
 PATH_PREFIX = "/_matrix/client/unstable/org.matrix.msc4140/delayed_events"
@@ -36,21 +21,6 @@ _HS_NAME = "red"
 _EVENT_TYPE = "com.example.test"
 
 
-class DelayedEventsUnstableSupportTestCase(HomeserverTestCase):
-    servlets = [versions.register_servlets]
-
-    def test_false_by_default(self) -> None:
-        channel = self.make_request("GET", "/_matrix/client/versions")
-        self.assertEqual(channel.code, 200, channel.result)
-        self.assertFalse(channel.json_body["unstable_features"]["org.matrix.msc4140"])
-
-    @unittest.override_config({"max_event_delay_duration": "24h"})
-    def test_true_if_enabled(self) -> None:
-        channel = self.make_request("GET", "/_matrix/client/versions")
-        self.assertEqual(channel.code, 200, channel.result)
-        self.assertTrue(channel.json_body["unstable_features"]["org.matrix.msc4140"])
-
-
 class DelayedEventsTestCase(HomeserverTestCase):
     """Tests getting and managing delayed events."""
 

tests/rest/client/test_rooms.py

@@ -4,7 +4,7 @@
 # Copyright 2019 The Matrix.org Foundation C.I.C.
 # Copyright 2017 Vector Creations Ltd
 # Copyright 2014-2016 OpenMarket Ltd
-# Copyright (C) 2023-2024 New Vector, Ltd
+# Copyright (C) 2023 New Vector, Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as

tests/rest/client/test_sync.py

@@ -282,33 +282,22 @@ class SyncTypingTests(unittest.HomeserverTestCase):
         self.assertEqual(200, channel.code)
         next_batch = channel.json_body["next_batch"]
 
+        # This should time out! But it does not, because our stream token is
+        # ahead, and therefore it's saying the typing (that we've actually
+        # already seen) is new, since it's got a token above our new, now-reset
+        # stream token.
+        channel = self.make_request("GET", sync_url % (access_token, next_batch))
+        self.assertEqual(200, channel.code)
+        next_batch = channel.json_body["next_batch"]
+
         # Clear the typing information, so that it doesn't think everything is
-        # in the future. This happens automatically when the typing stream
-        # resets.
+        # in the future.
         typing._reset()
 
-        # Nothing new, so we time out.
+        # Now it SHOULD fail as it never completes!
         with self.assertRaises(TimedOutException):
             self.make_request("GET", sync_url % (access_token, next_batch))
 
-        # Sync and start typing again.
-        sync_channel = self.make_request(
-            "GET", sync_url % (access_token, next_batch), await_result=False
-        )
-        self.assertFalse(sync_channel.is_finished())
-
-        channel = self.make_request(
-            "PUT",
-            typing_url % (room, other_user_id, other_access_token),
-            b'{"typing": true, "timeout": 30000}',
-        )
-        self.assertEqual(200, channel.code)
-
-        # Sync should now return.
-        sync_channel.await_result()
-        self.assertEqual(200, sync_channel.code)
-        next_batch = sync_channel.json_body["next_batch"]
-
 
 class SyncKnockTestCase(KnockingStrippedStateEventHelperMixin):
     servlets = [

tests/util/test_stream_change_cache.py

@@ -53,8 +53,8 @@ class StreamChangeCacheTests(unittest.HomeserverTestCase):
         # return True, whether it's a known entity or not.
         self.assertTrue(cache.has_entity_changed("user@foo.com", 0))
         self.assertTrue(cache.has_entity_changed("not@here.website", 0))
-        self.assertTrue(cache.has_entity_changed("user@foo.com", 2))
-        self.assertTrue(cache.has_entity_changed("not@here.website", 2))
+        self.assertTrue(cache.has_entity_changed("user@foo.com", 3))
+        self.assertTrue(cache.has_entity_changed("not@here.website", 3))
 
     def test_entity_has_changed_pops_off_start(self) -> None:
         """
@@ -76,11 +76,9 @@ class StreamChangeCacheTests(unittest.HomeserverTestCase):
         self.assertTrue("user@foo.com" not in cache._entity_to_key)
 
         self.assertEqual(
-            cache.get_all_entities_changed(2).entities,
-            ["bar@baz.net", "user@elsewhere.org"],
+            cache.get_all_entities_changed(3).entities, ["user@elsewhere.org"]
         )
-        self.assertFalse(cache.get_all_entities_changed(1).hit)
-        self.assertTrue(cache.get_all_entities_changed(2).hit)
+        self.assertFalse(cache.get_all_entities_changed(2).hit)
 
         # If we update an existing entity, it keeps the two existing entities
         cache.entity_has_changed("bar@baz.net", 5)
@@ -91,8 +89,7 @@ class StreamChangeCacheTests(unittest.HomeserverTestCase):
             cache.get_all_entities_changed(3).entities,
             ["user@elsewhere.org", "bar@baz.net"],
         )
-        self.assertFalse(cache.get_all_entities_changed(1).hit)
-        self.assertTrue(cache.get_all_entities_changed(2).hit)
+        self.assertFalse(cache.get_all_entities_changed(2).hit)
 
     def test_get_all_entities_changed(self) -> None:
         """
@@ -117,8 +114,7 @@ class StreamChangeCacheTests(unittest.HomeserverTestCase):
         self.assertEqual(
             cache.get_all_entities_changed(3).entities, ["user@elsewhere.org"]
         )
-        self.assertFalse(cache.get_all_entities_changed(0).hit)
-        self.assertTrue(cache.get_all_entities_changed(1).hit)
+        self.assertFalse(cache.get_all_entities_changed(1).hit)
 
         # ... later, things gest more updates
         cache.entity_has_changed("user@foo.com", 5)
@@ -153,7 +149,7 @@ class StreamChangeCacheTests(unittest.HomeserverTestCase):
         # With no entities, it returns True for the past, present, and False for
         # the future.
         self.assertTrue(cache.has_any_entity_changed(0))
-        self.assertFalse(cache.has_any_entity_changed(1))
+        self.assertTrue(cache.has_any_entity_changed(1))
         self.assertFalse(cache.has_any_entity_changed(2))
 
         # We add an entity