Repository: matrix-org/synapse

Issue:
`SerializationFailure` when rotating push actions
We recently changed things so we rotate notifications almost immediately, rather than waiting a day. This interacts badly with people sending read receipts, as that transaction attempts to modify the push summary tables.

We need to fix this or revert before we make another release.

Broke by https://github.com/matrix-org/synapse/pull/13005

Stack trace:

```
SerializationFailure: could not serialize access due to concurrent update

  File "synapse/metrics/background_process_metrics.py", line 240, in run
    return await func(*args, **kwargs)
  File "synapse/storage/databases/main/event_push_actions.py", line 806, in _rotate_notifs
    caught_up = await self.db_pool.runInteraction(
  File "synapse/storage/database.py", line 835, in runInteraction
    return await delay_cancellation(_runInteraction())
  File "twisted/internet/defer.py", line 1656, in _inlineCallbacks
    result = current_context.run(
  File "twisted/python/failure.py", line 514, in throwExceptionIntoGenerator
    return g.throw(self.type, self.value, self.tb)
  File "synapse/storage/database.py", line 807, in _runInteraction
    result = await self.runWithConnection(
  File "synapse/storage/database.py", line 930, in runWithConnection
    return await make_deferred_yieldable(
  File "twisted/python/threadpool.py", line 244, in inContext
    result = inContext.theWork()  # type: ignore[attr-defined]
  File "twisted/python/threadpool.py", line 260, in <lambda>
    inContext.theWork = lambda: context.call(  # type: ignore[attr-defined]
  File "twisted/python/context.py", line 117, in callWithContext
    return self.currentContext().callWithContext(ctx, func, *args, **kw)
  File "twisted/python/context.py", line 82, in callWithContext
    return func(*args, **kw)
  File "twisted/enterprise/adbapi.py", line 282, in _runWithConnection
    result = func(conn, *args, **kw)
  File "synapse/storage/database.py", line 923, in inner_func
    return func(db_conn, *args, **kwargs)
  File "synapse/storage/database.py", line 671, in new_transaction
    r = func(cursor, *args, **kwargs)
  File "synapse/storage/databases/main/event_push_actions.py", line 850, in _rotate_notifs_txn
    self._rotate_notifs_before_txn(txn, rotate_to_stream_ordering)
  File "synapse/storage/databases/main/event_push_actions.py", line 950, in _rotate_notifs_before_txn
    txn.execute_batch(
  File "synapse/storage/database.py", line 329, in execute_batch
    self._do_execute(
  File "synapse/storage/database.py", line 395, in _do_execute
    return func(sql, *args, **kwargs)
  File "synapse/storage/database.py", line 330, in <lambda>
    lambda the_sql: execute_batch(self.txn, the_sql, args), sql
  File "psycopg2/extras.py", line 1187, in execute_batch
    cur.execute(b";".join(sqls))
```

Sentry: https://sentry.tools.element.io/organizations/element/issues/924/

Discussion:
erikjohnston: > Major concern: if we're now only processing read receipts every 30 seconds, does that mean it takes 30 seconds for the unread indicator to disappear?

No, it means that for those 30s the `event_push_summary` table won't be used to calculate unread counts (which should be fine as there won't be any/many push actions to count after getting a receipt).

`get_unread_event_push_actions_by_room_for_user` works broadly by: a) querying `event_push_summary` for any usable counts, and then b) querying `event_push_actions` for actions that haven't been included in the summary. So in the first step won't return any counts, and so the second step will query `event_push_actions` for all push actions in the range from the receipt to now. 
richvdh: > `get_unread_event_push_actions_by_room_for_user` works broadly by: a) querying `event_push_summary` for any usable counts, and then b) querying `event_push_actions` for actions that haven't been included in the summary. So in the first step won't return any counts, and so the second step will query `event_push_actions` for all push actions in the range from the receipt to now.

ok, makes sense. it would be great to include that sort of explanation in the comments in the code!

Patch:
diff --git a/changelog.d/13118.misc b/changelog.d/13118.misc
new file mode 100644
index 000000000..3bb51962e
--- /dev/null
+++ b/changelog.d/13118.misc
@@ -0,0 +1,1 @@
+Reduce DB usage of `/sync` when a large number of unread messages have recently been sent in a room.
diff --git a/changelog.d/13123.removal b/changelog.d/13123.removal
new file mode 100644
index 000000000..f013f1616
--- /dev/null
+++ b/changelog.d/13123.removal
@@ -0,0 +1,1 @@
+Remove the unspecced `DELETE /directory/list/room/{roomId}` endpoint, which hid rooms from the [public room directory](https://spec.matrix.org/v1.3/client-server-api/#listing-rooms). Instead, `PUT` to the same URL with a visibility of `"private"`.
\ No newline at end of file
diff --git a/changelog.d/13124.misc b/changelog.d/13124.misc
new file mode 100644
index 000000000..513078f8d
--- /dev/null
+++ b/changelog.d/13124.misc
@@ -0,0 +1,1 @@
+Add type annotations to `tests.test_server`.
diff --git a/mypy.ini b/mypy.ini
index c5130feae..4b08f45c6 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -113,6 +113,9 @@ disallow_untyped_defs = False
 [mypy-tests.handlers.test_user_directory]
 disallow_untyped_defs = True
 
+[mypy-tests.test_server]
+disallow_untyped_defs = True
+
 [mypy-tests.state.test_profile]
 disallow_untyped_defs = True
 
diff --git a/synapse/rest/client/directory.py b/synapse/rest/client/directory.py
index d6c89cb16..bc1b18c92 100644
--- a/synapse/rest/client/directory.py
+++ b/synapse/rest/client/directory.py
@@ -151,17 +151,6 @@ class ClientDirectoryListServer(RestServlet):
 
         return 200, {}
 
-    async def on_DELETE(
-        self, request: SynapseRequest, room_id: str
-    ) -> Tuple[int, JsonDict]:
-        requester = await self.auth.get_user_by_req(request)
-
-        await self.directory_handler.edit_published_room_list(
-            requester, room_id, "private"
-        )
-
-        return 200, {}
-
 
 class ClientAppserviceDirectoryListServer(RestServlet):
     PATTERNS = client_patterns(
diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py
index 10a796238..80ca2fd0b 100644
--- a/synapse/storage/databases/main/event_push_actions.py
+++ b/synapse/storage/databases/main/event_push_actions.py
@@ -233,14 +233,30 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, EventsWorkerStore, SQLBas
 
         counts = NotifCounts()
 
-        # First we pull the counts from the summary table
+        # First we pull the counts from the summary table.
+        #
+        # We check that `last_receipt_stream_ordering` matches the stream
+        # ordering given. If it doesn't match then a new read receipt has arrived and
+        # we haven't yet updated the counts in `event_push_summary` to reflect
+        # that; in that case we simply ignore `event_push_summary` counts
+        # and do a manual count of all of the rows in the `event_push_actions` table
+        # for this user/room.
+        #
+        # If `last_receipt_stream_ordering` is null then that means it's up to
+        # date (as the row was written by an older version of Synapse that
+        # updated `event_push_summary` synchronously when persisting a new read
+        # receipt).
         txn.execute(
             """
                 SELECT stream_ordering, notif_count, COALESCE(unread_count, 0)
                 FROM event_push_summary
-                WHERE room_id = ? AND user_id = ? AND stream_ordering > ?
+                WHERE room_id = ? AND user_id = ?
+                AND (
+                    (last_receipt_stream_ordering IS NULL AND stream_ordering > ?)
+                    OR last_receipt_stream_ordering = ?
+                )
             """,
-            (room_id, user_id, stream_ordering),
+            (room_id, user_id, stream_ordering, stream_ordering),
         )
         row = txn.fetchone()
 
@@ -263,9 +279,9 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, EventsWorkerStore, SQLBas
         if row:
             counts.highlight_count += row[0]
 
-        # Finally we need to count push actions that haven't been summarized
-        # yet.
-        # We only want to pull out push actions that we haven't summarized yet.
+        # Finally we need to count push actions that aren't included in the
+        # summary returned above, e.g. recent events that haven't been
+        # summarized yet, or the summary is empty due to a recent read receipt.
         stream_ordering = max(stream_ordering, summary_stream_ordering)
         notify_count, unread_count = self._get_notif_unread_count_for_user_room(
             txn, room_id, user_id, stream_ordering
@@ -800,6 +816,19 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, EventsWorkerStore, SQLBas
         self._doing_notif_rotation = True
 
         try:
+            # First we recalculate push summaries and delete stale push actions
+            # for rooms/users with new receipts.
+            while True:
+                logger.debug("Handling new receipts")
+
+                caught_up = await self.db_pool.runInteraction(
+                    "_handle_new_receipts_for_notifs_txn",
+                    self._handle_new_receipts_for_notifs_txn,
+                )
+                if caught_up:
+                    break
+
+            # Then we update the event push summaries for any new events
             while True:
                 logger.info("Rotating notifications")
 
@@ -810,10 +839,110 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, EventsWorkerStore, SQLBas
                     break
                 await self.hs.get_clock().sleep(self._rotate_delay)
 
+            # Finally we clear out old event push actions.
             await self._remove_old_push_actions_that_have_rotated()
         finally:
             self._doing_notif_rotation = False
 
+    def _handle_new_receipts_for_notifs_txn(self, txn: LoggingTransaction) -> bool:
+        """Check for new read receipts and delete from event push actions.
+
+        Any push actions which predate the user's most recent read receipt are
+        now redundant, so we can remove them from `event_push_actions` and
+        update `event_push_summary`.
+        """
+
+        limit = 100
+
+        min_stream_id = self.db_pool.simple_select_one_onecol_txn(
+            txn,
+            table="event_push_summary_last_receipt_stream_id",
+            keyvalues={},
+            retcol="stream_id",
+        )
+
+        sql = """
+            SELECT r.stream_id, r.room_id, r.user_id, e.stream_ordering
+            FROM receipts_linearized AS r
+            INNER JOIN events AS e USING (event_id)
+            WHERE r.stream_id > ? AND user_id LIKE ?
+            ORDER BY r.stream_id ASC
+            LIMIT ?
+        """
+
+        # We only want local users, so we add a dodgy filter to the above query
+        # and recheck it below.
+        user_filter = "%:" + self.hs.hostname
+
+        txn.execute(
+            sql,
+            (
+                min_stream_id,
+                user_filter,
+                limit,
+            ),
+        )
+        rows = txn.fetchall()
+
+        # For each new read receipt we delete push actions from before it and
+        # recalculate the summary.
+        for _, room_id, user_id, stream_ordering in rows:
+            # Only handle our own read receipts.
+            if not self.hs.is_mine_id(user_id):
+                continue
+
+            txn.execute(
+                """
+                DELETE FROM event_push_actions
+                WHERE room_id = ?
+                    AND user_id = ?
+                    AND stream_ordering <= ?
+                    AND highlight = 0
+                """,
+                (room_id, user_id, stream_ordering),
+            )
+
+            old_rotate_stream_ordering = self.db_pool.simple_select_one_onecol_txn(
+                txn,
+                table="event_push_summary_stream_ordering",
+                keyvalues={},
+                retcol="stream_ordering",
+            )
+
+            notif_count, unread_count = self._get_notif_unread_count_for_user_room(
+                txn, room_id, user_id, stream_ordering, old_rotate_stream_ordering
+            )
+
+            self.db_pool.simple_upsert_txn(
+                txn,
+                table="event_push_summary",
+                keyvalues={"room_id": room_id, "user_id": user_id},
+                values={
+                    "notif_count": notif_count,
+                    "unread_count": unread_count,
+                    "stream_ordering": old_rotate_stream_ordering,
+                    "last_receipt_stream_ordering": stream_ordering,
+                },
+            )
+
+        # We always update `event_push_summary_last_receipt_stream_id` to
+        # ensure that we don't rescan the same receipts for remote users.
+        #
+        # This requires repeatable read to be safe, as we need the
+        # `MAX(stream_id)` to not include any new rows that have been committed
+        # since the start of the transaction (since those rows won't have been
+        # returned by the query above). Alternatively we could query the max
+        # stream ID at the start of the transaction and bound everything by
+        # that.
+        txn.execute(
+            """
+            UPDATE event_push_summary_last_receipt_stream_id
+            SET stream_id = (SELECT COALESCE(MAX(stream_id), 0) FROM receipts_linearized)
+            """
+        )
+
+        return len(rows) < limit
+
     def _rotate_notifs_txn(self, txn: LoggingTransaction) -> bool:
         """Archives older notifications into event_push_summary. Returns whether
         the archiving process has caught up or not.
@@ -1033,66 +1162,6 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, EventsWorkerStore, SQLBas
             if done:
                 break
 
-    def _remove_old_push_actions_before_txn(
-        self, txn: LoggingTransaction, room_id: str, user_id: str, stream_ordering: int
-    ) -> None:
-        """
-        Purges old push actions for a user and room before a given
-        stream_ordering.
-
-        We however keep a months worth of highlighted notifications, so that
-        users can still get a list of recent highlights.
-
-        Args:
-            txn: The transaction
-            room_id: Room ID to delete from
-            user_id: user ID to delete for
-            stream_ordering: The lowest stream ordering which will
-                                  not be deleted.
-        """
-        txn.call_after(
-            self.get_unread_event_push_actions_by_room_for_user.invalidate,
-            (room_id, user_id),
-        )
-
-        # We need to join on the events table to get the received_ts for
-        # event_push_actions and sqlite won't let us use a join in a delete so
-        # we can't just delete where received_ts < x. Furthermore we can
-        # only identify event_push_actions by a tuple of room_id, event_id
-        # we we can't use a subquery.
-        # Instead, we look up the stream ordering for the last event in that
-        # room received before the threshold time and delete event_push_actions
-        # in the room with a stream_odering before that.
-        txn.execute(
-            "DELETE FROM event_push_actions "
-            " WHERE user_id = ? AND room_id = ? AND "
-            " stream_ordering <= ?"
-            " AND ((stream_ordering < ? AND highlight = 1) or highlight = 0)",
-            (user_id, room_id, stream_ordering, self.stream_ordering_month_ago),
-        )
-
-        old_rotate_stream_ordering = self.db_pool.simple_select_one_onecol_txn(
-            txn,
-            table="event_push_summary_stream_ordering",
-            keyvalues={},
-            retcol="stream_ordering",
-        )
-
-        notif_count, unread_count = self._get_notif_unread_count_for_user_room(
-            txn, room_id, user_id, stream_ordering, old_rotate_stream_ordering
-        )
-
-        self.db_pool.simple_upsert_txn(
-            txn,
-            table="event_push_summary",
-            keyvalues={"room_id": room_id, "user_id": user_id},
-            values={
-                "notif_count": notif_count,
-                "unread_count": unread_count,
-                "stream_ordering": old_rotate_stream_ordering,
-            },
-        )
-
 
 class EventPushActionsStore(EventPushActionsWorkerStore):
     EPA_HIGHLIGHT_INDEX = "epa_highlight_index"
diff --git a/synapse/storage/databases/main/receipts.py b/synapse/storage/databases/main/receipts.py
index bec6d6057..0090c9f22 100644
--- a/synapse/storage/databases/main/receipts.py
+++ b/synapse/storage/databases/main/receipts.py
@@ -26,7 +26,7 @@ from typing import (
     cast,
 )
 
-from synapse.api.constants import EduTypes, ReceiptTypes
+from synapse.api.constants import EduTypes
 from synapse.replication.slave.storage._slaved_id_tracker import SlavedIdTracker
 from synapse.replication.tcp.streams import ReceiptsStream
 from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
@@ -682,17 +682,6 @@ class ReceiptsWorkerStore(SQLBaseStore):
             lock=False,
         )
 
-        # When updating a local users read receipt, remove any push actions
-        # which resulted from the receipt's event and all earlier events.
-        if (
-            self.hs.is_mine_id(user_id)
-            and receipt_type in (ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE)
-            and stream_ordering is not None
-        ):
-            self._remove_old_push_actions_before_txn(  # type: ignore[attr-defined]
-                txn, room_id=room_id, user_id=user_id, stream_ordering=stream_ordering
-            )
-
         return rx_ts
 
     def _graph_to_linear(
diff --git a/synapse/storage/schema/main/delta/72/01event_push_summary_receipt.sql b/synapse/storage/schema/main/delta/72/01event_push_summary_receipt.sql
new file mode 100644
index 000000000..e45db6152
--- /dev/null
+++ b/synapse/storage/schema/main/delta/72/01event_push_summary_receipt.sql
@@ -0,0 +1,35 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- Add a column that records the position of the read receipt for the user at
+-- the time we summarised the push actions. This is used to check if the counts
+-- are up to date after a new read receipt has been sent.
+--
+-- Null means that we can skip that check, as the row was written by an older
+-- version of Synapse that updated `event_push_summary` synchronously when
+-- persisting a new read receipt
+ALTER TABLE event_push_summary ADD COLUMN last_receipt_stream_ordering BIGINT;
+
+
+-- Tracks which new receipts we've handled
+CREATE TABLE event_push_summary_last_receipt_stream_id (
+    Lock CHAR(1) NOT NULL DEFAULT 'X' UNIQUE,  -- Makes sure this table only has one row.
+    stream_id BIGINT NOT NULL,
+    CHECK (Lock='X')
+);
+
+INSERT INTO event_push_summary_last_receipt_stream_id (stream_id)
+  SELECT COALESCE(MAX(stream_id), 0)
+  FROM receipts_linearized;


Test patch:
diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py
index 2ac5f6db5..ef069a811 100644
--- a/tests/storage/test_event_push_actions.py
+++ b/tests/storage/test_event_push_actions.py
@@ -55,7 +55,7 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
 
     def test_count_aggregation(self) -> None:
         room_id = "!foo:example.com"
-        user_id = "@user1235:example.com"
+        user_id = "@user1235:test"
 
         last_read_stream_ordering = [0]
 
@@ -81,11 +81,26 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
         def _inject_actions(stream: int, action: list) -> None:
             event = Mock()
             event.room_id = room_id
-            event.event_id = "$test:example.com"
+            event.event_id = f"$test{stream}:example.com"
             event.internal_metadata.stream_ordering = stream
             event.internal_metadata.is_outlier.return_value = False
             event.depth = stream
 
+            self.get_success(
+                self.store.db_pool.simple_insert(
+                    table="events",
+                    values={
+                        "stream_ordering": stream,
+                        "topological_ordering": stream,
+                        "type": "m.room.message",
+                        "room_id": room_id,
+                        "processed": True,
+                        "outlier": False,
+                        "event_id": event.event_id,
+                    },
+                )
+            )
+
             self.get_success(
                 self.store.add_push_actions_to_staging(
                     event.event_id,
@@ -105,18 +120,28 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
         def _rotate(stream: int) -> None:
             self.get_success(
                 self.store.db_pool.runInteraction(
-                    "", self.store._rotate_notifs_before_txn, stream
+                    "rotate-receipts", self.store._handle_new_receipts_for_notifs_txn
+                )
+            )
+
+            self.get_success(
+                self.store.db_pool.runInteraction(
+                    "rotate-notifs", self.store._rotate_notifs_before_txn, stream
                 )
             )
 
         def _mark_read(stream: int, depth: int) -> None:
             last_read_stream_ordering[0] = stream
+
             self.get_success(
                 self.store.db_pool.runInteraction(
                     "",
-                    self.store._remove_old_push_actions_before_txn,
+                    self.store._insert_linearized_receipt_txn,
                     room_id,
+                    "m.read",
                     user_id,
+                    f"$test{stream}:example.com",
+                    {},
                     stream,
                 )
             )
@@ -150,7 +175,7 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
 
         _assert_counts(1, 0)
 
-        _mark_read(7, 7)
+        _mark_read(6, 6)
         _assert_counts(0, 0)
 
         _inject_actions(8, HIGHLIGHT)
diff --git a/tests/test_server.py b/tests/test_server.py
index 847432f79..fc4bce899 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -14,7 +14,7 @@
 
 import re
 from http import HTTPStatus
-from typing import Tuple
+from typing import Awaitable, Callable, Dict, NoReturn, Optional, Tuple
 
 from twisted.internet.defer import Deferred
 from twisted.web.resource import Resource
@@ -36,6 +36,7 @@ from synapse.util import Clock
 from tests import unittest
 from tests.http.server._base import test_disconnect
 from tests.server import (
+    FakeChannel,
     FakeSite,
     ThreadedMemoryReactorClock,
     make_request,
@@ -44,7 +45,7 @@ from tests.server import (
 
 
 class JsonResourceTests(unittest.TestCase):
-    def setUp(self):
+    def setUp(self) -> None:
         self.reactor = ThreadedMemoryReactorClock()
         self.hs_clock = Clock(self.reactor)
         self.homeserver = setup_test_homeserver(
@@ -54,7 +55,7 @@ class JsonResourceTests(unittest.TestCase):
             reactor=self.reactor,
         )
 
-    def test_handler_for_request(self):
+    def test_handler_for_request(self) -> None:
         """
         JsonResource.handler_for_request gives correctly decoded URL args to
         the callback, while Twisted will give the raw bytes of URL query
@@ -62,7 +63,9 @@ class JsonResourceTests(unittest.TestCase):
         """
         got_kwargs = {}
 
-        def _callback(request, **kwargs):
+        def _callback(
+            request: SynapseRequest, **kwargs: object
+        ) -> Tuple[int, Dict[str, object]]:
             got_kwargs.update(kwargs)
             return 200, kwargs
 
@@ -83,13 +86,13 @@ class JsonResourceTests(unittest.TestCase):
 
         self.assertEqual(got_kwargs, {"room_id": "\N{SNOWMAN}"})
 
-    def test_callback_direct_exception(self):
+    def test_callback_direct_exception(self) -> None:
         """
         If the web callback raises an uncaught exception, it will be translated
         into a 500.
         """
 
-        def _callback(request, **kwargs):
+        def _callback(request: SynapseRequest, **kwargs: object) -> NoReturn:
             raise Exception("boo")
 
         res = JsonResource(self.homeserver)
@@ -103,17 +106,17 @@ class JsonResourceTests(unittest.TestCase):
 
         self.assertEqual(channel.result["code"], b"500")
 
-    def test_callback_indirect_exception(self):
+    def test_callback_indirect_exception(self) -> None:
         """
         If the web callback raises an uncaught exception in a Deferred, it will
         be translated into a 500.
         """
 
-        def _throw(*args):
+        def _throw(*args: object) -> NoReturn:
             raise Exception("boo")
 
-        def _callback(request, **kwargs):
-            d = Deferred()
+        def _callback(request: SynapseRequest, **kwargs: object) -> "Deferred[None]":
+            d: "Deferred[None]" = Deferred()
             d.addCallback(_throw)
             self.reactor.callLater(0.5, d.callback, True)
             return make_deferred_yieldable(d)
@@ -129,13 +132,13 @@ class JsonResourceTests(unittest.TestCase):
 
         self.assertEqual(channel.result["code"], b"500")
 
-    def test_callback_synapseerror(self):
+    def test_callback_synapseerror(self) -> None:
         """
         If the web callback raises a SynapseError, it returns the appropriate
         status code and message set in it.
         """
 
-        def _callback(request, **kwargs):
+        def _callback(request: SynapseRequest, **kwargs: object) -> NoReturn:
             raise SynapseError(403, "Forbidden!!one!", Codes.FORBIDDEN)
 
         res = JsonResource(self.homeserver)
@@ -151,12 +154,12 @@ class JsonResourceTests(unittest.TestCase):
         self.assertEqual(channel.json_body["error"], "Forbidden!!one!")
         self.assertEqual(channel.json_body["errcode"], "M_FORBIDDEN")
 
-    def test_no_handler(self):
+    def test_no_handler(self) -> None:
         """
         If there is no handler to process the request, Synapse will return 400.
         """
 
-        def _callback(request, **kwargs):
+        def _callback(request: SynapseRequest, **kwargs: object) -> None:
             """
             Not ever actually called!
             """
@@ -175,14 +178,16 @@ class JsonResourceTests(unittest.TestCase):
         self.assertEqual(channel.json_body["error"], "Unrecognized request")
         self.assertEqual(channel.json_body["errcode"], "M_UNRECOGNIZED")
 
-    def test_head_request(self):
+    def test_head_request(self) -> None:
         """
         JsonResource.handler_for_request gives correctly decoded URL args to
         the callback, while Twisted will give the raw bytes of URL query
         arguments.
         """
 
-        def _callback(request, **kwargs):
+        def _callback(
+            request: SynapseRequest, **kwargs: object
+        ) -> Tuple[int, Dict[str, object]]:
             return 200, {"result": True}
 
         res = JsonResource(self.homeserver)
@@ -203,20 +208,21 @@ class JsonResourceTests(unittest.TestCase):
 
 
 class OptionsResourceTests(unittest.TestCase):
-    def setUp(self):
+    def setUp(self) -> None:
         self.reactor = ThreadedMemoryReactorClock()
 
         class DummyResource(Resource):
             isLeaf = True
 
-            def render(self, request):
-                return request.path
+            def render(self, request: SynapseRequest) -> bytes:
+                # Type-ignore: mypy thinks request.path is Optional[Any], not bytes.
+                return request.path  # type: ignore[return-value]
 
         # Setup a resource with some children.
         self.resource = OptionsResource()
         self.resource.putChild(b"res", DummyResource())
 
-    def _make_request(self, method, path):
+    def _make_request(self, method: bytes, path: bytes) -> FakeChannel:
         """Create a request from the method/path and return a channel with the response."""
         # Create a site and query for the resource.
         site = SynapseSite(
@@ -233,7 +239,7 @@ class OptionsResourceTests(unittest.TestCase):
         channel = make_request(self.reactor, site, method, path, shorthand=False)
         return channel
 
-    def test_unknown_options_request(self):
+    def test_unknown_options_request(self) -> None:
         """An OPTIONS requests to an unknown URL still returns 204 No Content."""
         channel = self._make_request(b"OPTIONS", b"/foo/")
         self.assertEqual(channel.result["code"], b"204")
@@ -253,7 +259,7 @@ class OptionsResourceTests(unittest.TestCase):
             "has CORS Headers header",
         )
 
-    def test_known_options_request(self):
+    def test_known_options_request(self) -> None:
         """An OPTIONS requests to an known URL still returns 204 No Content."""
         channel = self._make_request(b"OPTIONS", b"/res/")
         self.assertEqual(channel.result["code"], b"204")
@@ -273,12 +279,12 @@ class OptionsResourceTests(unittest.TestCase):
             "has CORS Headers header",
         )
 
-    def test_unknown_request(self):
+    def test_unknown_request(self) -> None:
         """A non-OPTIONS request to an unknown URL should 404."""
         channel = self._make_request(b"GET", b"/foo/")
         self.assertEqual(channel.result["code"], b"404")
 
-    def test_known_request(self):
+    def test_known_request(self) -> None:
         """A non-OPTIONS request to an known URL should query the proper resource."""
         channel = self._make_request(b"GET", b"/res/")
         self.assertEqual(channel.result["code"], b"200")
@@ -287,16 +293,17 @@ class OptionsResourceTests(unittest.TestCase):
 
 class WrapHtmlRequestHandlerTests(unittest.TestCase):
     class TestResource(DirectServeHtmlResource):
-        callback = None
+        callback: Optional[Callable[..., Awaitable[None]]]
 
-        async def _async_render_GET(self, request):
+        async def _async_render_GET(self, request: SynapseRequest) -> None:
+            assert self.callback is not None
             await self.callback(request)
 
-    def setUp(self):
+    def setUp(self) -> None:
         self.reactor = ThreadedMemoryReactorClock()
 
-    def test_good_response(self):
-        async def callback(request):
+    def test_good_response(self) -> None:
+        async def callback(request: SynapseRequest) -> None:
             request.write(b"response")
             request.finish()
 
@@ -311,13 +318,13 @@ class WrapHtmlRequestHandlerTests(unittest.TestCase):
         body = channel.result["body"]
         self.assertEqual(body, b"response")
 
-    def test_redirect_exception(self):
+    def test_redirect_exception(self) -> None:
         """
         If the callback raises a RedirectException, it is turned into a 30x
         with the right location.
         """
 
-        async def callback(request, **kwargs):
+        async def callback(request: SynapseRequest, **kwargs: object) -> None:
             raise RedirectException(b"/look/an/eagle", 301)
 
         res = WrapHtmlRequestHandlerTests.TestResource()
@@ -332,13 +339,13 @@ class WrapHtmlRequestHandlerTests(unittest.TestCase):
         location_headers = [v for k, v in headers if k == b"Location"]
         self.assertEqual(location_headers, [b"/look/an/eagle"])
 
-    def test_redirect_exception_with_cookie(self):
+    def test_redirect_exception_with_cookie(self) -> None:
         """
         If the callback raises a RedirectException which sets a cookie, that is
         returned too
         """
 
-        async def callback(request, **kwargs):
+        async def callback(request: SynapseRequest, **kwargs: object) -> NoReturn:
             e = RedirectException(b"/no/over/there", 304)
             e.cookies.append(b"session=yespls")
             raise e
@@ -357,10 +364,10 @@ class WrapHtmlRequestHandlerTests(unittest.TestCase):
         cookies_headers = [v for k, v in headers if k == b"Set-Cookie"]
         self.assertEqual(cookies_headers, [b"session=yespls"])
 
-    def test_head_request(self):
+    def test_head_request(self) -> None:
         """A head request should work by being turned into a GET request."""
 
-        async def callback(request):
+        async def callback(request: SynapseRequest) -> None:
             request.write(b"response")
             request.finish()
 
@@ -410,7 +417,7 @@ class CancellableDirectServeHtmlResource(DirectServeHtmlResource):
 class DirectServeJsonResourceCancellationTests(unittest.TestCase):
     """Tests for `DirectServeJsonResource` cancellation."""
 
-    def setUp(self):
+    def setUp(self) -> None:
         self.reactor = ThreadedMemoryReactorClock()
         self.clock = Clock(self.reactor)
         self.resource = CancellableDirectServeJsonResource(self.clock)
@@ -444,7 +451,7 @@ class DirectServeJsonResourceCancellationTests(unittest.TestCase):
 class DirectServeHtmlResourceCancellationTests(unittest.TestCase):
     """Tests for `DirectServeHtmlResource` cancellation."""
 
-    def setUp(self):
+    def setUp(self) -> None:
         self.reactor = ThreadedMemoryReactorClock()
         self.clock = Clock(self.reactor)
         self.resource = CancellableDirectServeHtmlResource(self.clock)