diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
index 41503899a11..07d4f6dfcea 100644
--- a/.github/PULL_REQUEST_TEMPLATE.md
+++ b/.github/PULL_REQUEST_TEMPLATE.md
@@ -9,6 +9,5 @@
   - End with either a period (.) or an exclamation mark (!).
   - Start with a capital letter.
   - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry.
-* [ ] Pull request includes a [sign off](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#sign-off)
 * [ ] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct
   (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters))
diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml
index 679b76440ec..010bce863bc 100644
--- a/.github/workflows/docker.yml
+++ b/.github/workflows/docker.yml
@@ -11,7 +11,7 @@ on:
 permissions:
   contents: read
   packages: write
-
+  id-token: write # needed for signing the images with GitHub OIDC Token
 jobs:
   build:
     runs-on: ubuntu-latest
@@ -29,6 +29,9 @@ jobs:
       - name: Inspect builder
         run: docker buildx inspect
 
+      - name: Install Cosign
+        uses: sigstore/cosign-installer@v3.3.0
+
       - name: Checkout repository
         uses: actions/checkout@v4
 
@@ -68,6 +71,7 @@ jobs:
             type=pep440,pattern={{raw}}
 
       - name: Build and push all platforms
+        id: build-and-push
         uses: docker/build-push-action@v5
         with:
           push: true
@@ -82,3 +86,14 @@ jobs:
           # https://github.com/rust-lang/cargo/issues/10583
           build-args: |
             CARGO_NET_GIT_FETCH_WITH_CLI=true
+
+      - name: Sign the images with GitHub OIDC Token
+        env:
+          DIGEST: ${{ steps.build-and-push.outputs.digest }}
+          TAGS: ${{ steps.set-tag.outputs.tags }}
+        run: |
+          images=""
+          for tag in ${TAGS}; do
+            images+="${tag}@${DIGEST} "
+          done
+          cosign sign --yes ${images}
diff --git a/.github/workflows/docs-add-version-picker.yaml b/.github/workflows/docs-add-version-picker.yaml
deleted file mode 100644
index 717d5c85d3a..00000000000
--- a/.github/workflows/docs-add-version-picker.yaml
+++ /dev/null
@@ -1,90 +0,0 @@
-name: Add Version Picker (RUN ONCE)
-
-on:
-  workflow_dispatch:
-
-jobs:
-  add-version-picker:
-    name: Add Version Picker
-    runs-on: ubuntu-latest
-
-    steps:
-      - uses: actions/checkout@v4
-
-      - name: Configure Git
-        run: |
-          git config user.email "action@synapse.bot.com"
-          git config user.name "Action Bot"
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-
-      - name: Setup mdbook
-        uses: peaceiris/actions-mdbook@adeb05db28a0c0004681db83893d56c0388ea9ea # v1.2.0
-        with:
-          mdbook-version: '0.4.17'
-
-      - name: Copy files to release branches
-        run: |
-          for version in "v1.98" "v1.97" "v1.96" "v1.95" "v1.94" "v1.93" "v1.92" "v1.91" "v1.90" "v1.89" "v1.88" "v1.87" "v1.86" "v1.85" "v1.84" "v1.83" "v1.82" "v1.81" "v1.80" "v1.79" "v1.78" "v1.77" "v1.76" "v1.75" "v1.74" "v1.73" "v1.72" "v1.71" "v1.70" "v1.69" "v1.68" "v1.67" "v1.66" "v1.65" "v1.64" "v1.63" "v1.62" "v1.61" "v1.60" "v1.59" "v1.58" "v1.57" "v1.56" "v1.55" "v1.54" "v1.53" "v1.52" "v1.51" "v1.50" "v1.49" "v1.48" "v1.47" "v1.46" "v1.45" "v1.44" "v1.43" "v1.42" "v1.41" "v1.40" "v1.39" "v1.38" "v1.37"
-          do
-            git fetch
-            git checkout -b release-$version origin/release-$version
-
-            git checkout develop -- ./book.toml
-            git checkout develop -- ./docs/website_files/version-picker.js
-            git checkout develop -- ./docs/website_files/version-picker.css
-            git checkout develop -- ./docs/website_files/README.md
-
-            echo "window.SYNAPSE_VERSION = '$version';" > ./docs/website_files/version.js
-
-            # Adding version-picker element to index.hbs
-            awk '/<button id="search-toggle" class="icon-button" type="button" title="Search. \(Shortkey: s\)" aria-label="Toggle Searchbar" aria-expanded="false" aria-keyshortcuts="S" aria-controls="searchbar">/{
-                print; getline; print; getline; print; getline; print;
-                print "\
-                                  <div class=\"version-picker\">\n\
-                                      <div class=\"dropdown\">\n\
-                                          <div class=\"select\">\n\
-                                              <span></span>\n\
-                                              <i class=\"fa fa-chevron-down\"></i>\n\
-                                          </div>\n\
-                                          <input type=\"hidden\" name=\"version\">\n\
-                                          <ul class=\"dropdown-menu\">\n\
-                                              <!-- Versions will be added dynamically in version-picker.js -->\n\
-                                          </ul>\n\
-                                      </div>\n\
-                                  </div>\
-                ";
-                next
-            } 1' ./docs/website_files/theme/index.hbs > output.html && mv output.html ./docs/website_files/theme/index.hbs
-
-            git add ./book.toml ./docs/website_files/version-picker.js ./docs/website_files/version-picker.css ./docs/website_files/version.js ./docs/website_files/README.md ./docs/website_files/theme/index.hbs
-            git commit -m "Version picker added for $version docs"
-            git push
-          done
-
-      - name: Build docs for Github Pages
-        run: |
-          git fetch
-          git branch gh-pages origin/gh-pages
-
-          for version in "v1.98" "v1.97" "v1.96" "v1.95" "v1.94" "v1.93" "v1.92" "v1.91" "v1.90" "v1.89" "v1.88" "v1.87" "v1.86" "v1.85" "v1.84" "v1.83" "v1.82" "v1.81" "v1.80" "v1.79" "v1.78" "v1.77" "v1.76" "v1.75" "v1.74" "v1.73" "v1.72" "v1.71" "v1.70" "v1.69" "v1.68" "v1.67" "v1.66" "v1.65" "v1.64" "v1.63" "v1.62" "v1.61" "v1.60" "v1.59" "v1.58" "v1.57" "v1.56" "v1.55" "v1.54" "v1.53" "v1.52" "v1.51" "v1.50" "v1.49" "v1.48" "v1.47" "v1.46" "v1.45" "v1.44" "v1.43" "v1.42" "v1.41" "v1.40" "v1.39" "v1.38" "v1.37"
-          do
-            git checkout release-$version
-
-            mdbook build && cp book/welcome_and_overview.html book/index.html
-            mkdir ver-temp && cp -r book/* ver-temp/
-            rm -r ./book
-
-            git checkout gh-pages
-            rm -r $version
-            mv ver-temp $version
-
-            git add ./$version
-            git commit -m "Version picker deployed for $version docs to Github Pages"
-          done
-
-      - name: Push to gh-pages
-        run: |
-          git checkout gh-pages
-          git status
-          git push
\ No newline at end of file
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index f22ca5f7e60..7ce22befce7 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -12,10 +12,6 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  check-signoff:
-    if: "github.event_name == 'pull_request'"
-    uses: "matrix-org/backend-meta/.github/workflows/sign-off.yml@v2"
-
   # Job to detect what has changed so we don't run e.g. Rust checks on PRs that
   # don't modify Rust code.
   changes:
@@ -286,10 +282,26 @@ jobs:
       - check-schema-delta
       - check-lockfile
       - lint-clippy
+      - lint-clippy-nightly
       - lint-rustfmt
     runs-on: ubuntu-latest
     steps:
-      - run: "true"
+      - uses: matrix-org/done-action@v2
+        with:
+          needs: ${{ toJSON(needs) }}
+
+          # Various bits are skipped if there was no applicable changes.
+          skippable: |
+            check-sampleconfig
+            check-schema-delta
+            lint
+            lint-mypy
+            lint-newsfile
+            lint-pydantic
+            lint-clippy
+            lint-clippy-nightly
+            lint-rustfmt
+
 
   calculate-test-jobs:
     if: ${{ !cancelled() && !failure() }} # Allow previous steps to be skipped, but not fail
@@ -699,6 +711,7 @@ jobs:
       - complement
       - cargo-test
       - cargo-bench
+      - linting-done
     runs-on: ubuntu-latest
     steps:
       - uses: matrix-org/done-action@v2
@@ -706,7 +719,7 @@ jobs:
           needs: ${{ toJSON(needs) }}
 
           # Various bits are skipped if there was no applicable changes.
-          # The newsfile and signoff lint may be skipped on non PR builds.
+          # The newsfile lint may be skipped on non PR builds.
           skippable: |
             trial
             trial-olddeps
@@ -714,7 +727,6 @@ jobs:
             portdb
             export-data
             complement
-            check-signoff
             lint-newsfile
             cargo-test
             cargo-bench
diff --git a/.github/workflows/triage-incoming.yml b/.github/workflows/triage-incoming.yml
index a53d0eaaa72..7a369b77fe4 100644
--- a/.github/workflows/triage-incoming.yml
+++ b/.github/workflows/triage-incoming.yml
@@ -1,9 +1,8 @@
 name: Move new issues into the issue triage board
 
 on:
-  # issues:
-  #   types: [ opened ]
-  workflow_dispatch:
+  issues:
+    types: [ opened ]
 
 jobs:
   triage:
diff --git a/CHANGES.md b/CHANGES.md
index cac80ca9721..d6f3d7cf33a 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,3 +1,50 @@
+# Synapse 1.99.0rc1 (2024-01-09)
+
+### Features
+
+- Add [config options](https://element-hq.github.io/synapse/v1.99/usage/configuration/config_documentation.html#server_notices) to set the avatar and the topic of the server notices room, as well as the avatar of the server notices user. ([\#16679](https://github.com/matrix-org/synapse/issues/16679))
+- Add config option [`email.notif_delay_before_mail`](https://element-hq.github.io/synapse/v1.99/usage/configuration/config_documentation.html#email) to tweak the delay before an email is sent following a notification. ([\#16696](https://github.com/matrix-org/synapse/issues/16696))
+- Add new configuration option [`sentry.environment`](https://element-hq.github.io/synapse/v1.99/usage/configuration/config_documentation.html#sentry) for improved system monitoring. Contributed by @zeeshanrafiqrana. ([\#16738](https://github.com/matrix-org/synapse/issues/16738))
+- Filter out rooms from the room directory being served to other homeservers when those rooms block that homeserver by their Access Control Lists. ([\#16759](https://github.com/element-hq/synapse/issues/16759))
+
+### Bugfixes
+
+- Fix a long-standing bug where the signing keys generated by Synapse were world-readable. Contributed by Fabian Klemp. ([\#16740](https://github.com/matrix-org/synapse/issues/16740))
+- Fix email verification redirection. Contributed by Fadhlan Ridhwanallah. ([\#16761](https://github.com/element-hq/synapse/issues/16761))
+- Fixed a bug that prevented users from being queried by display name if it contains non-ASCII characters. ([\#16767](https://github.com/element-hq/synapse/issues/16767))
+- Allow reactivate user without password with Admin API in some edge cases. ([\#16770](https://github.com/element-hq/synapse/issues/16770))
+- Adds the `recursion_depth` parameter to the response of the /relations endpoint if MSC3981 recursion is being performed. ([\#16775](https://github.com/element-hq/synapse/issues/16775))
+
+### Improved Documentation
+
+- Added version picker for Synapse documentation. Contributed by @Dmytro27Ind. ([\#16533](https://github.com/matrix-org/synapse/issues/16533))
+- Clarify that `password_config.enabled: "only_for_reauth"` does not allow new logins to be created using password auth. ([\#16737](https://github.com/matrix-org/synapse/issues/16737))
+- Remove value from header in configuration documentation for `refresh_token_lifetime`. ([\#16763](https://github.com/element-hq/synapse/issues/16763))
+- Add another custom statistics collection server to the documentation. Contributed by @loelkes. ([\#16769](https://github.com/element-hq/synapse/issues/16769))
+
+### Internal Changes
+
+- Remove run-once workflow after adding the version picker to the documentation. ([\#9453](https://github.com/element-hq/synapse/issues/9453))
+- Update the implementation of [MSC2965](https://github.com/matrix-org/matrix-spec-proposals/pull/2965) (OIDC Provider discovery). ([\#16726](https://github.com/matrix-org/synapse/issues/16726))
+- Move the rust stubs inline for better IDE integration. ([\#16757](https://github.com/element-hq/synapse/issues/16757))
+- Fix sample config doc CI. ([\#16758](https://github.com/element-hq/synapse/issues/16758))
+- Simplify event internal metadata class. ([\#16762](https://github.com/element-hq/synapse/issues/16762), [\#16780](https://github.com/element-hq/synapse/issues/16780))
+- Sign the published docker image using [cosign](https://docs.sigstore.dev/). ([\#16774](https://github.com/element-hq/synapse/issues/16774))
+- Port `EventInternalMetadata` class to Rust. ([\#16782](https://github.com/element-hq/synapse/issues/16782))
+
+
+
+### Updates to locked dependencies
+
+* Bump actions/setup-go from 4 to 5. ([\#16749](https://github.com/matrix-org/synapse/issues/16749))
+* Bump actions/setup-python from 4 to 5. ([\#16748](https://github.com/matrix-org/synapse/issues/16748))
+* Bump immutabledict from 3.0.0 to 4.0.0. ([\#16743](https://github.com/matrix-org/synapse/issues/16743))
+* Bump isort from 5.12.0 to 5.13.0. ([\#16745](https://github.com/matrix-org/synapse/issues/16745))
+* Bump isort from 5.13.0 to 5.13.1. ([\#16752](https://github.com/matrix-org/synapse/issues/16752))
+* Bump pydantic from 2.5.1 to 2.5.2. ([\#16747](https://github.com/matrix-org/synapse/issues/16747))
+* Bump ruff from 0.1.6 to 0.1.7. ([\#16746](https://github.com/matrix-org/synapse/issues/16746))
+* Bump types-setuptools from 68.2.0.2 to 69.0.0.0. ([\#16744](https://github.com/matrix-org/synapse/issues/16744))
+
 # Synapse 1.98.0 (2023-12-12)
 
 Synapse 1.98.0 will be the last Synapse release in 2023; the regular release cadence will resume in January 2024.
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 2c85edf7127..1bdaef41c36 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -1,3 +1,3 @@
 # Welcome to Synapse
 
-Please see the [contributors' guide](https://matrix-org.github.io/synapse/latest/development/contributing_guide.html) in our rendered documentation.
+Please see the [contributors' guide](https://element-hq.github.io/synapse/latest/development/contributing_guide.html) in our rendered documentation.
diff --git a/Cargo.lock b/Cargo.lock
index d5e77297f4a..c60eacbf3d9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -13,9 +13,9 @@ dependencies = [
 
 [[package]]
 name = "anyhow"
-version = "1.0.75"
+version = "1.0.79"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4668cab20f66d8d020e1fbc0ebe47217433c1b6c8f2040faf858554e394ace6"
+checksum = "080e9890a082662b09c1ad45f567faeeb47f22b5fb23895fbe1e651e718e25ca"
 
 [[package]]
 name = "arc-swap"
@@ -188,9 +188,9 @@ dependencies = [
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.64"
+version = "1.0.76"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78803b62cbf1f46fde80d7c0e803111524b9877184cfe7c3033659490ac7a7da"
+checksum = "95fc56cda0b5c3325f5fbbd7ff9fda9e02bb00bb3dac51252d2f1bfa1cb8cc8c"
 dependencies = [
  "unicode-ident",
 ]
@@ -280,9 +280,9 @@ dependencies = [
 
 [[package]]
 name = "quote"
-version = "1.0.29"
+version = "1.0.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "573015e8ab27661678357f27dc26460738fd2b6c86e46f386fde94cb5d913105"
+checksum = "291ec9ab5efd934aaf503a6466c5d5251535d108ee747472c3977cc5acc868ef"
 dependencies = [
  "proc-macro2",
 ]
@@ -339,18 +339,18 @@ checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"
 
 [[package]]
 name = "serde"
-version = "1.0.193"
+version = "1.0.195"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "25dd9975e68d0cb5aa1120c288333fc98731bd1dd12f561e468ea4728c042b89"
+checksum = "63261df402c67811e9ac6def069e4786148c4563f4b50fd4bf30aa370d626b02"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.193"
+version = "1.0.195"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43576ca501357b9b071ac53cdc7da8ef0cbd9493d8df094cd821777ea6e894d3"
+checksum = "46fe8f8603d81ba86327b23a2e9cdf49e1255fb94a4c5f297f6ee0547178ea2c"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -359,9 +359,9 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.108"
+version = "1.0.111"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3d1c7e3eac408d115102c4c24ad393e0821bb3a5df4d506a80f85f7a742a526b"
+checksum = "176e46fa42316f18edd598015a5166857fc835ec732f5215eac6b7bdbf0a84f4"
 dependencies = [
  "itoa",
  "ryu",
@@ -382,9 +382,9 @@ checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601"
 
 [[package]]
 name = "syn"
-version = "2.0.28"
+version = "2.0.48"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "04361975b3f5e348b2189d8dc55bc942f278b2d482a6a0365de5bdd62d351567"
+checksum = "0f3531638e407dfc0814761abb7c00a5b54992b849452a0646b7f65c9f770f3f"
 dependencies = [
  "proc-macro2",
  "quote",
diff --git a/changelog.d/16533.doc b/changelog.d/16533.doc
deleted file mode 100644
index ae23a8a5782..00000000000
--- a/changelog.d/16533.doc
+++ /dev/null
@@ -1 +0,0 @@
-Added version picker for Synapse documentation. Contributed by @Dmytro27Ind.
\ No newline at end of file
diff --git a/changelog.d/16679.feature b/changelog.d/16679.feature
deleted file mode 100644
index 85af837ae17..00000000000
--- a/changelog.d/16679.feature
+++ /dev/null
@@ -1 +0,0 @@
-Add config options to set the avatar and the topic of the server notices room.
diff --git a/changelog.d/16696.feature b/changelog.d/16696.feature
deleted file mode 100644
index 53d7b40f36e..00000000000
--- a/changelog.d/16696.feature
+++ /dev/null
@@ -1 +0,0 @@
-Add a setting to be able to tweak the delay without interaction before an email is sent following a notification.
diff --git a/changelog.d/16726.misc b/changelog.d/16726.misc
deleted file mode 100644
index bac312465ca..00000000000
--- a/changelog.d/16726.misc
+++ /dev/null
@@ -1 +0,0 @@
-Update the implementation of [MSC2965](https://github.com/matrix-org/matrix-spec-proposals/pull/2965) (OIDC Provider discovery).
diff --git a/changelog.d/16737.doc b/changelog.d/16737.doc
deleted file mode 100644
index 26035b73ec3..00000000000
--- a/changelog.d/16737.doc
+++ /dev/null
@@ -1 +0,0 @@
-Clarify that `password_config.enabled: "only_for_reauth"` does not allow new logins to be created using password auth.
\ No newline at end of file
diff --git a/changelog.d/16738.feature b/changelog.d/16738.feature
deleted file mode 100644
index c9ea12a2abc..00000000000
--- a/changelog.d/16738.feature
+++ /dev/null
@@ -1 +0,0 @@
-Add new Sentry configuration option `environment` for improved system monitoring. Contributed by @zeeshanrafiqrana.
\ No newline at end of file
diff --git a/changelog.d/16740.bugfix b/changelog.d/16740.bugfix
deleted file mode 100644
index 21551516e2d..00000000000
--- a/changelog.d/16740.bugfix
+++ /dev/null
@@ -1 +0,0 @@
-Fix a long-standing bug where the signing keys generated by Synapse were world-readable. Contributed by Fabian Klemp.
diff --git a/changelog.d/16757.misc b/changelog.d/16757.misc
deleted file mode 100644
index 9856bf1aa67..00000000000
--- a/changelog.d/16757.misc
+++ /dev/null
@@ -1 +0,0 @@
-Move the rust stubs inline for better IDE integration.
diff --git a/changelog.d/16758.misc b/changelog.d/16758.misc
deleted file mode 100644
index eea2d0869c5..00000000000
--- a/changelog.d/16758.misc
+++ /dev/null
@@ -1 +0,0 @@
-Fix sample config doc CI.
diff --git a/changelog.d/16766.misc b/changelog.d/16766.misc
new file mode 100644
index 00000000000..ded77a11c40
--- /dev/null
+++ b/changelog.d/16766.misc
@@ -0,0 +1 @@
+Split up deleting devices into batches.
diff --git a/changelog.d/16776.misc b/changelog.d/16776.misc
new file mode 100644
index 00000000000..1650075a77e
--- /dev/null
+++ b/changelog.d/16776.misc
@@ -0,0 +1 @@
+Remove CI check for sign off as we require an CLA signature instead.
diff --git a/changelog.d/16778.doc b/changelog.d/16778.doc
new file mode 100644
index 00000000000..fe3ca0da7ef
--- /dev/null
+++ b/changelog.d/16778.doc
@@ -0,0 +1 @@
+Add a link to the "Request log format" explainer on the "Logging sample config" documentation page.
diff --git a/changelog.d/16781.misc b/changelog.d/16781.misc
new file mode 100644
index 00000000000..2f628dc5cbd
--- /dev/null
+++ b/changelog.d/16781.misc
@@ -0,0 +1 @@
+Ensure CI fails when linting fails to make sure auto-merge does the correct thing.
diff --git a/changelog.d/16785.misc b/changelog.d/16785.misc
new file mode 100644
index 00000000000..4de185c5dd7
--- /dev/null
+++ b/changelog.d/16785.misc
@@ -0,0 +1 @@
+Reduce amount of state pulled out when querying federation hierachy.
diff --git a/changelog.d/16788.misc b/changelog.d/16788.misc
new file mode 100644
index 00000000000..e58a5a7a320
--- /dev/null
+++ b/changelog.d/16788.misc
@@ -0,0 +1 @@
+Pull less state out of the DB when we retry fetching old events during backfill.
diff --git a/changelog.d/7.misc b/changelog.d/7.misc
new file mode 100644
index 00000000000..63f1fb77ffa
--- /dev/null
+++ b/changelog.d/7.misc
@@ -0,0 +1 @@
+Faster partial join to room with complex auth graph.
diff --git a/debian/changelog b/debian/changelog
index abcfedf6d65..cb183146a83 100644
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,9 @@
+matrix-synapse-py3 (1.99.0~rc1) stable; urgency=medium
+
+  * New Synapse release 1.99.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 09 Jan 2024 13:43:56 +0000
+
 matrix-synapse-py3 (1.98.0) stable; urgency=medium
 
   * New Synapse release 1.98.0.
diff --git a/docs/admin_api/user_admin_api.md b/docs/admin_api/user_admin_api.md
index e8e492d0951..9dc600b8753 100644
--- a/docs/admin_api/user_admin_api.md
+++ b/docs/admin_api/user_admin_api.md
@@ -149,10 +149,11 @@ Body parameters:
   granting them access to the Admin API, among other things.
 - `deactivated` - **bool**, optional. If unspecified, deactivation state will be left unchanged.
 
-  Note: the `password` field must also be set if both of the following are true:
-  - `deactivated` is set to `false` and the user was previously deactivated (you are reactivating this user)
-  - Users are allowed to set their password on this homeserver (both `password_config.enabled` and
-    `password_config.localdb_enabled` config options are set to `true`).
+  Note:
+  - For the password field there is no strict check of the necessity for its presence.
+    It is possible to have active users without a password, e.g. when authenticating with OIDC is configured.
+    You must check yourself whether a password is required when reactivating a user or not.
+  - It is not possible to set a password if the config option `password_config.localdb_enabled` is set `false`.
   Users' passwords are wiped upon account deactivation, hence the need to set a new one here.
 
   Note: a user cannot be erased with this API. For more details on
@@ -223,7 +224,7 @@ The following parameters should be set in the URL:
   **or** displaynames that contain this value.
 - `guests` - string representing a bool - Is optional and if `false` will **exclude** guest users.
   Defaults to `true` to include guest users. This parameter is not supported when MSC3861 is enabled. [See #15582](https://github.com/matrix-org/synapse/pull/15582)
-- `admins` - Optional flag to filter admins. If `true`, only admins are queried. If `false`, admins are excluded from 
+- `admins` - Optional flag to filter admins. If `true`, only admins are queried. If `false`, admins are excluded from
   the query. When the flag is absent (the default), **both** admins and non-admins are included in the search results.
 - `deactivated` - string representing a bool - Is optional and if `true` will **include** deactivated users.
   Defaults to `false` to exclude deactivated users.
@@ -272,7 +273,7 @@ The following fields are returned in the JSON response body:
   - `is_guest` - bool - Status if that user is a guest account.
   - `admin` - bool - Status if that user is a server administrator.
   - `user_type` - string - Type of the user. Normal users are type `None`.
-    This allows user type specific behaviour. There are also types `support` and `bot`. 
+    This allows user type specific behaviour. There are also types `support` and `bot`.
   - `deactivated` - bool - Status if that user has been marked as deactivated.
   - `erased` - bool - Status if that user has been marked as erased.
   - `shadow_banned` - bool - Status if that user has been marked as shadow banned.
@@ -887,7 +888,7 @@ The following fields are returned in the JSON response body:
 
 ### Create a device
 
-Creates a new device for a specific `user_id` and `device_id`. Does nothing if the `device_id` 
+Creates a new device for a specific `user_id` and `device_id`. Does nothing if the `device_id`
 exists already.
 
 The API is:
@@ -1254,11 +1255,11 @@ The following parameters should be set in the URL:
 
 ## Check username availability
 
-Checks to see if a username is available, and valid, for the server. See [the client-server 
+Checks to see if a username is available, and valid, for the server. See [the client-server
 API](https://matrix.org/docs/spec/client_server/r0.6.0#get-matrix-client-r0-register-available)
 for more information.
 
-This endpoint will work even if registration is disabled on the server, unlike 
+This endpoint will work even if registration is disabled on the server, unlike
 `/_matrix/client/r0/register/available`.
 
 The API is:
diff --git a/docs/development/contributing_guide.md b/docs/development/contributing_guide.md
index e575ceda1f2..08de320b1bd 100644
--- a/docs/development/contributing_guide.md
+++ b/docs/development/contributing_guide.md
@@ -4,16 +4,16 @@ This document aims to get you started with contributing to Synapse!
 
 # 1. Who can contribute to Synapse?
 
-Everyone is welcome to contribute code to [Synapse](https://github.com/element-hq/synapse),
-provided that they are willing to
-license their contributions under the same license as the project itself. We
-follow a simple 'inbound=outbound' model for contributions: the act of
-submitting an 'inbound' contribution means that the contributor agrees to
-license the code under the same terms as the project's overall 'outbound'
-license - in our case, this is almost always Apache Software License v2 (see
-[LICENSE](https://github.com/element-hq/synapse/blob/develop/LICENSE)).
+Everyone is welcome to contribute code to
+[Synapse](https://github.com/element-hq/synapse), provided that they are willing
+to license their contributions to Element under a [Contributor License
+Agreement](https://cla-assistant.io/element-hq/synapse) (CLA). This ensures that
+their contribution will be made available under an OSI-approved open-source
+license, currently Affero General Public License v3 (AGPLv3).
 
-TODO THIS NEEDS UPDATING
+Please see the
+[Element blog post](https://element.io/blog/synapse-now-lives-at-github-com-element-hq-synapse/)
+for the full rationale.
 
 # 2. What do I need?
 
@@ -499,81 +499,19 @@ separate pull requests.)
 
 ## Sign off
 
-In order to have a concrete record that your contribution is intentional
-and you agree to license it under the same terms as the project's license, we've adopted the
-same lightweight approach that the Linux Kernel
-[submitting patches process](
-https://www.kernel.org/doc/html/latest/process/submitting-patches.html#sign-your-work-the-developer-s-certificate-of-origin>),
-[Docker](https://github.com/docker/docker/blob/master/CONTRIBUTING.md), and many other
-projects use: the DCO ([Developer Certificate of Origin](http://developercertificate.org/)).
-This is a simple declaration that you wrote
-the contribution or otherwise have the right to contribute it to Matrix:
+After you make a PR a comment from @CLAassistant will appear asking you to sign
+the [CLA](https://cla-assistant.io/element-hq/synapse).
+This will link a page to allow you to confirm that you have read and agreed to
+the CLA by signing in with GitHub.
 
-```
-Developer Certificate of Origin
-Version 1.1
-
-Copyright (C) 2004, 2006 The Linux Foundation and its contributors.
-660 York Street, Suite 102,
-San Francisco, CA 94110 USA
-
-Everyone is permitted to copy and distribute verbatim copies of this
-license document, but changing it is not allowed.
-
-Developer's Certificate of Origin 1.1
-
-By making a contribution to this project, I certify that:
-
-(a) The contribution was created in whole or in part by me and I
-    have the right to submit it under the open source license
-    indicated in the file; or
-
-(b) The contribution is based upon previous work that, to the best
-    of my knowledge, is covered under an appropriate open source
-    license and I have the right under that license to submit that
-    work with modifications, whether created in whole or in part
-    by me, under the same open source license (unless I am
-    permitted to submit under a different license), as indicated
-    in the file; or
-
-(c) The contribution was provided directly to me by some other
-    person who certified (a), (b) or (c) and I have not modified
-    it.
-
-(d) I understand and agree that this project and the contribution
-    are public and that a record of the contribution (including all
-    personal information I submit with it, including my sign-off) is
-    maintained indefinitely and may be redistributed consistent with
-    this project or the open source license(s) involved.
-```
-
-If you agree to this for your contribution, then all that's needed is to
-include the line in your commit or pull request comment:
-
-```
-Signed-off-by: Your Name <your@email.example.org>
-```
+Alternatively, you can sign off before opening a PR by going to
+<https://cla-assistant.io/element-hq/synapse>.
 
 We accept contributions under a legally identifiable name, such as
 your name on government documentation or common-law names (names
 claimed by legitimate usage or repute). Unfortunately, we cannot
 accept anonymous contributions at this time.
 
-Git allows you to add this signoff automatically when using the `-s`
-flag to `git commit`, which uses the name and email set in your
-`user.name` and `user.email` git configs.
-
-### Private Sign off
-
-If you would like to provide your legal name privately to the Matrix.org
-Foundation (instead of in a public commit or comment), you can do so
-by emailing your legal name and a link to the pull request to
-[dco@matrix.org](mailto:dco@matrix.org?subject=Private%20sign%20off).
-It helps to include "sign off" or similar in the subject line. You will then
-be instructed further.
-
-Once private sign off is complete, doing so for future contributions will not
-be required.
 
 # 10. Turn feedback into better code.
 
diff --git a/docs/usage/administration/monitoring/reporting_homeserver_usage_statistics.md b/docs/usage/administration/monitoring/reporting_homeserver_usage_statistics.md
index 60b758e33b0..4c0dbb5acd0 100644
--- a/docs/usage/administration/monitoring/reporting_homeserver_usage_statistics.md
+++ b/docs/usage/administration/monitoring/reporting_homeserver_usage_statistics.md
@@ -74,3 +74,4 @@ consider using one of the following known implementations:
 
 * [Matrix.org's Panopticon](https://github.com/matrix-org/panopticon)
 * [Famedly's Barad-dûr](https://gitlab.com/famedly/infra/services/barad-dur)
+* [Synapse Usage Exporter](https://github.com/loelkes/synapse-usage-exporter) for Prometheus
diff --git a/docs/usage/configuration/config_documentation.md b/docs/usage/configuration/config_documentation.md
index 3e49b8a2991..8723b9a3feb 100644
--- a/docs/usage/configuration/config_documentation.md
+++ b/docs/usage/configuration/config_documentation.md
@@ -2680,7 +2680,7 @@ Example configuration:
 refreshable_access_token_lifetime: 10m
 ```
 ---
-### `refresh_token_lifetime: 24h`
+### `refresh_token_lifetime`
 
 Time that a refresh token remains valid for (provided that it is not
 exchanged for another one first).
diff --git a/docs/usage/configuration/logging_sample_config.md b/docs/usage/configuration/logging_sample_config.md
index 89567419973..23a55abdccd 100644
--- a/docs/usage/configuration/logging_sample_config.md
+++ b/docs/usage/configuration/logging_sample_config.md
@@ -11,6 +11,9 @@ Note that a default logging configuration (shown below) is created automatically
 the homeserver config when following the [installation instructions](../../setup/installation.md).
 It should be named `<SERVERNAME>.log.config` by default.
 
+Hint: If you're looking for a guide on what each of the fields in the "Processed request" log lines mean,
+see [Request log format](../administration/request_log.md).
+
 ```yaml
 {{#include ../../sample_log_config.yaml}}
 ```
diff --git a/poetry.lock b/poetry.lock
index b3e046b52da..80ff6de5d99 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -64,17 +64,17 @@ tests-no-zope = ["cloudpickle", "hypothesis", "mypy (>=1.1.1)", "pympler", "pyte
 
 [[package]]
 name = "authlib"
-version = "1.2.1"
+version = "1.3.0"
 description = "The ultimate Python library in building OAuth and OpenID Connect servers and clients."
 optional = true
-python-versions = "*"
+python-versions = ">=3.8"
 files = [
-    {file = "Authlib-1.2.1-py2.py3-none-any.whl", hash = "sha256:c88984ea00149a90e3537c964327da930779afa4564e354edfd98410bea01911"},
-    {file = "Authlib-1.2.1.tar.gz", hash = "sha256:421f7c6b468d907ca2d9afede256f068f87e34d23dd221c07d13d4c234726afb"},
+    {file = "Authlib-1.3.0-py2.py3-none-any.whl", hash = "sha256:9637e4de1fb498310a56900b3e2043a206b03cb11c05422014b0302cbc814be3"},
+    {file = "Authlib-1.3.0.tar.gz", hash = "sha256:959ea62a5b7b5123c5059758296122b57cd2585ae2ed1c0622c21b371ffdae06"},
 ]
 
 [package.dependencies]
-cryptography = ">=3.2"
+cryptography = "*"
 
 [[package]]
 name = "automat"
@@ -2475,13 +2475,13 @@ doc = ["Sphinx", "sphinx-rtd-theme"]
 
 [[package]]
 name = "sentry-sdk"
-version = "1.35.0"
+version = "1.39.1"
 description = "Python client for Sentry (https://sentry.io)"
 optional = true
 python-versions = "*"
 files = [
-    {file = "sentry-sdk-1.35.0.tar.gz", hash = "sha256:04e392db9a0d59bd49a51b9e3a92410ac5867556820465057c2ef89a38e953e9"},
-    {file = "sentry_sdk-1.35.0-py2.py3-none-any.whl", hash = "sha256:a7865952701e46d38b41315c16c075367675c48d049b90a4cc2e41991ebc7efa"},
+    {file = "sentry-sdk-1.39.1.tar.gz", hash = "sha256:320a55cdf9da9097a0bead239c35b7e61f53660ef9878861824fd6d9b2eaf3b5"},
+    {file = "sentry_sdk-1.39.1-py2.py3-none-any.whl", hash = "sha256:81b5b9ffdd1a374e9eb0c053b5d2012155db9cbe76393a8585677b753bd5fdc1"},
 ]
 
 [package.dependencies]
@@ -3048,13 +3048,13 @@ files = [
 
 [[package]]
 name = "types-jsonschema"
-version = "4.20.0.0"
+version = "4.20.0.20240105"
 description = "Typing stubs for jsonschema"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "types-jsonschema-4.20.0.0.tar.gz", hash = "sha256:0de1032d243f1d3dba8b745ad84efe8c1af71665a9deb1827636ac535dcb79c1"},
-    {file = "types_jsonschema-4.20.0.0-py3-none-any.whl", hash = "sha256:e6d5df18aaca4412f0aae246a294761a92040e93d7bc840f002b7329a8b72d26"},
+    {file = "types-jsonschema-4.20.0.20240105.tar.gz", hash = "sha256:4a71af7e904498e7ad055149f6dc1eee04153b59a99ad7dd17aa3769c9bc5982"},
+    {file = "types_jsonschema-4.20.0.20240105-py3-none-any.whl", hash = "sha256:26706cd70a273e59e718074c4e756608a25ba61327a7f9a4493ebd11941e5ad4"},
 ]
 
 [package.dependencies]
diff --git a/pyproject.toml b/pyproject.toml
index 2ac03edfe80..24791aa0a97 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -96,7 +96,7 @@ module-name = "synapse.synapse_rust"
 
 [tool.poetry]
 name = "matrix-synapse"
-version = "1.98.0"
+version = "1.99.0rc1"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "AGPL-3.0-or-later"
diff --git a/rust/src/events/internal_metadata.rs b/rust/src/events/internal_metadata.rs
new file mode 100644
index 00000000000..a53601862dc
--- /dev/null
+++ b/rust/src/events/internal_metadata.rs
@@ -0,0 +1,430 @@
+/*
+ * 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>.
+ *
+ * Originally licensed under the Apache License, Version 2.0:
+ * <http://www.apache.org/licenses/LICENSE-2.0>.
+ *
+ * [This file includes modifications made by New Vector Limited]
+ *
+ */
+
+//! Implements the internal metadata class attached to events.
+//!
+//! The internal metadata is a bit like a `TypedDict`, in that it is stored as a
+//! JSON dict in the DB. Most events have zero, or only a few, of these keys
+//! set. Therefore, since we care more about memory size than performance here,
+//! we store these fields in a mapping.
+//!
+//! We want to store (most) of the fields as Rust objects, so we implement the
+//! mapping by using a vec of enums. This is less efficient than using
+//! attributes, but for small number of keys is actually faster than using a
+//! hash or btree map.
+
+use std::{num::NonZeroI64, ops::Deref};
+
+use anyhow::Context;
+use log::warn;
+use pyo3::{
+    exceptions::PyAttributeError,
+    pyclass, pymethods,
+    types::{PyDict, PyString},
+    IntoPy, PyAny, PyObject, PyResult, Python,
+};
+
+/// Definitions of the various fields of the internal metadata.
+#[derive(Clone)]
+enum EventInternalMetadataData {
+    OutOfBandMembership(bool),
+    SendOnBehalfOf(Box<str>),
+    RecheckRedaction(bool),
+    SoftFailed(bool),
+    ProactivelySend(bool),
+    Redacted(bool),
+    TxnId(Box<str>),
+    TokenId(i64),
+    DeviceId(Box<str>),
+}
+
+impl EventInternalMetadataData {
+    /// Convert the field to its name and python object.
+    fn to_python_pair<'a>(&self, py: Python<'a>) -> (&'a PyString, PyObject) {
+        match self {
+            EventInternalMetadataData::OutOfBandMembership(o) => {
+                (pyo3::intern!(py, "out_of_band_membership"), o.into_py(py))
+            }
+            EventInternalMetadataData::SendOnBehalfOf(o) => {
+                (pyo3::intern!(py, "send_on_behalf_of"), o.into_py(py))
+            }
+            EventInternalMetadataData::RecheckRedaction(o) => {
+                (pyo3::intern!(py, "recheck_redaction"), o.into_py(py))
+            }
+            EventInternalMetadataData::SoftFailed(o) => {
+                (pyo3::intern!(py, "soft_failed"), o.into_py(py))
+            }
+            EventInternalMetadataData::ProactivelySend(o) => {
+                (pyo3::intern!(py, "proactively_send"), o.into_py(py))
+            }
+            EventInternalMetadataData::Redacted(o) => {
+                (pyo3::intern!(py, "redacted"), o.into_py(py))
+            }
+            EventInternalMetadataData::TxnId(o) => (pyo3::intern!(py, "txn_id"), o.into_py(py)),
+            EventInternalMetadataData::TokenId(o) => (pyo3::intern!(py, "token_id"), o.into_py(py)),
+            EventInternalMetadataData::DeviceId(o) => {
+                (pyo3::intern!(py, "device_id"), o.into_py(py))
+            }
+        }
+    }
+
+    /// Converts from python key/values to the field.
+    ///
+    /// Returns `None` if the key is a valid but unrecognized string.
+    fn from_python_pair(key: &PyAny, value: &PyAny) -> PyResult<Option<Self>> {
+        let key_str: &str = key.extract()?;
+
+        let e = match key_str {
+            "out_of_band_membership" => EventInternalMetadataData::OutOfBandMembership(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+
+            "send_on_behalf_of" => EventInternalMetadataData::SendOnBehalfOf(
+                value
+                    .extract()
+                    .map(String::into_boxed_str)
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+            "recheck_redaction" => EventInternalMetadataData::RecheckRedaction(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+            "soft_failed" => EventInternalMetadataData::SoftFailed(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+            "proactively_send" => EventInternalMetadataData::ProactivelySend(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+            "redacted" => EventInternalMetadataData::Redacted(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+            "txn_id" => EventInternalMetadataData::TxnId(
+                value
+                    .extract()
+                    .map(String::into_boxed_str)
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+            "token_id" => EventInternalMetadataData::TokenId(
+                value
+                    .extract()
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+            "device_id" => EventInternalMetadataData::DeviceId(
+                value
+                    .extract()
+                    .map(String::into_boxed_str)
+                    .with_context(|| format!("'{key_str}' has invalid type"))?,
+            ),
+            _ => return Ok(None),
+        };
+
+        Ok(Some(e))
+    }
+}
+
+/// Helper macro to find the given field in internal metadata, returning None if
+/// not found.
+macro_rules! get_property_opt {
+    ($self:expr, $name:ident) => {
+        $self.data.iter().find_map(|entry| {
+            if let EventInternalMetadataData::$name(data) = entry {
+                Some(data)
+            } else {
+                None
+            }
+        })
+    };
+}
+
+/// Helper macro to find the given field in internal metadata, raising an
+/// attribute error if not found.
+macro_rules! get_property {
+    ($self:expr, $name:ident) => {
+        get_property_opt!($self, $name).ok_or_else(|| {
+            PyAttributeError::new_err(format!(
+                "'EventInternalMetadata' has no attribute '{}'",
+                stringify!($name),
+            ))
+        })
+    };
+}
+
+/// Helper macro to set the give field.
+macro_rules! set_property {
+    ($self:expr, $name:ident, $obj:expr) => {
+        for entry in &mut $self.data {
+            if let EventInternalMetadataData::$name(data) = entry {
+                *data = $obj;
+                return;
+            }
+        }
+
+        $self.data.push(EventInternalMetadataData::$name($obj))
+    };
+}
+
+#[pyclass]
+#[derive(Clone)]
+pub struct EventInternalMetadata {
+    /// The fields of internal metadata. This functions as a mapping.
+    data: Vec<EventInternalMetadataData>,
+
+    /// The stream ordering of this event. None, until it has been persisted.
+    #[pyo3(get, set)]
+    stream_ordering: Option<NonZeroI64>,
+
+    /// whether this event is an outlier (ie, whether we have the state at that
+    /// point in the DAG)
+    #[pyo3(get, set)]
+    outlier: bool,
+}
+
+#[pymethods]
+impl EventInternalMetadata {
+    #[new]
+    fn new(dict: &PyDict) -> PyResult<Self> {
+        let mut data = Vec::with_capacity(dict.len());
+
+        for (key, value) in dict.iter() {
+            match EventInternalMetadataData::from_python_pair(key, value) {
+                Ok(Some(entry)) => data.push(entry),
+                Ok(None) => {}
+                Err(err) => {
+                    warn!("Ignoring internal metadata field '{key}', as failed to convert to Rust due to {err}")
+                }
+            }
+        }
+
+        data.shrink_to_fit();
+
+        Ok(EventInternalMetadata {
+            data,
+            stream_ordering: None,
+            outlier: false,
+        })
+    }
+
+    fn copy(&self) -> Self {
+        self.clone()
+    }
+
+    fn get_dict(&self, py: Python<'_>) -> PyResult<PyObject> {
+        let dict = PyDict::new(py);
+
+        for entry in &self.data {
+            let (key, value) = entry.to_python_pair(py);
+            dict.set_item(key, value)?;
+        }
+
+        Ok(dict.into())
+    }
+
+    fn is_outlier(&self) -> bool {
+        self.outlier
+    }
+
+    /// Whether this event is an out-of-band membership.
+    ///
+    /// OOB memberships are a special case of outlier events: they are
+    /// membership events for federated rooms that we aren't full members of.
+    /// Examples include invites received over federation, and rejections for
+    /// such invites.
+    ///
+    /// The concept of an OOB membership is needed because these events need to
+    /// be processed as if they're new regular events (e.g. updating membership
+    /// state in the database, relaying to clients via /sync, etc) despite being
+    /// outliers.
+    ///
+    /// See also
+    /// https://element-hq.github.io/synapse/develop/development/room-dag-concepts.html#out-of-band-membership-events.
+    ///
+    /// (Added in synapse 0.99.0, so may be unreliable for events received
+    /// before that)
+    fn is_out_of_band_membership(&self) -> bool {
+        get_property_opt!(self, OutOfBandMembership)
+            .copied()
+            .unwrap_or(false)
+    }
+
+    /// Whether this server should send the event on behalf of another server.
+    /// This is used by the federation "send_join" API to forward the initial
+    /// join event for a server in the room.
+    ///
+    /// returns a str with the name of the server this event is sent on behalf
+    /// of.
+    fn get_send_on_behalf_of(&self) -> Option<&str> {
+        let s = get_property_opt!(self, SendOnBehalfOf);
+        s.map(|a| a.deref())
+    }
+
+    /// Whether the redaction event needs to be rechecked when fetching
+    /// from the database.
+    ///
+    /// Starting in room v3 redaction events are accepted up front, and later
+    /// checked to see if the redacter and redactee's domains match.
+    ///
+    /// If the sender of the redaction event is allowed to redact any event
+    /// due to auth rules, then this will always return false.
+    fn need_to_check_redaction(&self) -> bool {
+        get_property_opt!(self, RecheckRedaction)
+            .copied()
+            .unwrap_or(false)
+    }
+
+    /// Whether the event has been soft failed.
+    ///
+    /// Soft failed events should be handled as usual, except:
+    /// 1. They should not go down sync or event streams, or generally sent to
+    ///    clients.
+    /// 2. They should not be added to the forward extremities (and therefore
+    ///    not to current state).
+    fn is_soft_failed(&self) -> bool {
+        get_property_opt!(self, SoftFailed)
+            .copied()
+            .unwrap_or(false)
+    }
+
+    /// Whether the event, if ours, should be sent to other clients and servers.
+    ///
+    /// This is used for sending dummy events internally. Servers and clients
+    /// can still explicitly fetch the event.
+    fn should_proactively_send(&self) -> bool {
+        get_property_opt!(self, ProactivelySend)
+            .copied()
+            .unwrap_or(true)
+    }
+
+    /// Whether the event has been redacted.
+    ///
+    /// This is used for efficiently checking whether an event has been marked
+    /// as redacted without needing to make another database call.
+    fn is_redacted(&self) -> bool {
+        get_property_opt!(self, Redacted).copied().unwrap_or(false)
+    }
+
+    /// Whether this event can trigger a push notification
+    fn is_notifiable(&self) -> bool {
+        !self.outlier || self.is_out_of_band_membership()
+    }
+
+    // ** The following are the getters and setters of the various properties **
+
+    #[getter]
+    fn get_out_of_band_membership(&self) -> PyResult<bool> {
+        let bool = get_property!(self, OutOfBandMembership)?;
+        Ok(*bool)
+    }
+    #[setter]
+    fn set_out_of_band_membership(&mut self, obj: bool) {
+        set_property!(self, OutOfBandMembership, obj);
+    }
+
+    #[getter(send_on_behalf_of)]
+    fn getter_send_on_behalf_of(&self) -> PyResult<&str> {
+        let s = get_property!(self, SendOnBehalfOf)?;
+        Ok(s)
+    }
+    #[setter]
+    fn set_send_on_behalf_of(&mut self, obj: String) {
+        set_property!(self, SendOnBehalfOf, obj.into_boxed_str());
+    }
+
+    #[getter]
+    fn get_recheck_redaction(&self) -> PyResult<bool> {
+        let bool = get_property!(self, RecheckRedaction)?;
+        Ok(*bool)
+    }
+    #[setter]
+    fn set_recheck_redaction(&mut self, obj: bool) {
+        set_property!(self, RecheckRedaction, obj);
+    }
+
+    #[getter]
+    fn get_soft_failed(&self) -> PyResult<bool> {
+        let bool = get_property!(self, SoftFailed)?;
+        Ok(*bool)
+    }
+    #[setter]
+    fn set_soft_failed(&mut self, obj: bool) {
+        set_property!(self, SoftFailed, obj);
+    }
+
+    #[getter]
+    fn get_proactively_send(&self) -> PyResult<bool> {
+        let bool = get_property!(self, ProactivelySend)?;
+        Ok(*bool)
+    }
+    #[setter]
+    fn set_proactively_send(&mut self, obj: bool) {
+        set_property!(self, ProactivelySend, obj);
+    }
+
+    #[getter]
+    fn get_redacted(&self) -> PyResult<bool> {
+        let bool = get_property!(self, Redacted)?;
+        Ok(*bool)
+    }
+    #[setter]
+    fn set_redacted(&mut self, obj: bool) {
+        set_property!(self, Redacted, obj);
+    }
+
+    /// The transaction ID, if it was set when the event was created.
+    #[getter]
+    fn get_txn_id(&self) -> PyResult<&str> {
+        let s = get_property!(self, TxnId)?;
+        Ok(s)
+    }
+    #[setter]
+    fn set_txn_id(&mut self, obj: String) {
+        set_property!(self, TxnId, obj.into_boxed_str());
+    }
+
+    /// The access token ID of the user who sent this event, if any.
+    #[getter]
+    fn get_token_id(&self) -> PyResult<i64> {
+        let r = get_property!(self, TokenId)?;
+        Ok(*r)
+    }
+    #[setter]
+    fn set_token_id(&mut self, obj: i64) {
+        set_property!(self, TokenId, obj);
+    }
+
+    /// The device ID of the user who sent this event, if any.
+    #[getter]
+    fn get_device_id(&self) -> PyResult<&str> {
+        let s = get_property!(self, DeviceId)?;
+        Ok(s)
+    }
+    #[setter]
+    fn set_device_id(&mut self, obj: String) {
+        set_property!(self, DeviceId, obj.into_boxed_str());
+    }
+}
diff --git a/rust/src/events/mod.rs b/rust/src/events/mod.rs
new file mode 100644
index 00000000000..ee857b3d72a
--- /dev/null
+++ b/rust/src/events/mod.rs
@@ -0,0 +1,41 @@
+/*
+ * 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>.
+ *
+ * Originally licensed under the Apache License, Version 2.0:
+ * <http://www.apache.org/licenses/LICENSE-2.0>.
+ *
+ * [This file includes modifications made by New Vector Limited]
+ *
+ */
+
+//! Classes for representing Events.
+
+use pyo3::{types::PyModule, PyResult, Python};
+
+mod internal_metadata;
+
+/// Called when registering modules with python.
+pub fn register_module(py: Python<'_>, m: &PyModule) -> PyResult<()> {
+    let child_module = PyModule::new(py, "events")?;
+    child_module.add_class::<internal_metadata::EventInternalMetadata>()?;
+
+    m.add_submodule(child_module)?;
+
+    // We need to manually add the module to sys.modules to make `from
+    // synapse.synapse_rust import events` work.
+    py.import("sys")?
+        .getattr("modules")?
+        .set_item("synapse.synapse_rust.events", child_module)?;
+
+    Ok(())
+}
diff --git a/rust/src/lib.rs b/rust/src/lib.rs
index c44c09bda71..7b3b579e550 100644
--- a/rust/src/lib.rs
+++ b/rust/src/lib.rs
@@ -3,6 +3,7 @@ use pyo3::prelude::*;
 use pyo3_log::ResetHandle;
 
 pub mod acl;
+pub mod events;
 pub mod push;
 
 lazy_static! {
@@ -41,6 +42,7 @@ fn synapse_rust(py: Python<'_>, m: &PyModule) -> PyResult<()> {
 
     acl::register_module(py, m)?;
     push::register_module(py, m)?;
+    events::register_module(py, m)?;
 
     Ok(())
 }
diff --git a/synapse/events/__init__.py b/synapse/events/__init__.py
index 80e17f0fb0d..92b406e336a 100644
--- a/synapse/events/__init__.py
+++ b/synapse/events/__init__.py
@@ -42,7 +42,8 @@
 
 from synapse.api.constants import RelationTypes
 from synapse.api.room_versions import EventFormatVersions, RoomVersion, RoomVersions
-from synapse.types import JsonDict, RoomStreamToken, StrCollection
+from synapse.synapse_rust.events import EventInternalMetadata
+from synapse.types import JsonDict, StrCollection
 from synapse.util.caches import intern_dict
 from synapse.util.frozenutils import freeze
 from synapse.util.stringutils import strtobool
@@ -74,7 +75,7 @@
 #
 # Note that DictProperty/DefaultDictProperty cannot actually be used with
 # EventBuilder as it lacks a _dict property.
-_DictPropertyInstance = Union["_EventInternalMetadata", "EventBase", "EventBuilder"]
+_DictPropertyInstance = Union["EventBase", "EventBuilder"]
 
 
 class DictProperty(Generic[T]):
@@ -111,7 +112,7 @@ def __get__(
         if instance is None:
             return self
         try:
-            assert isinstance(instance, (EventBase, _EventInternalMetadata))
+            assert isinstance(instance, EventBase)
             return instance._dict[self.key]
         except KeyError as e1:
             # We want this to look like a regular attribute error (mostly so that
@@ -127,11 +128,11 @@ def __get__(
             ) from e1.__context__
 
     def __set__(self, instance: _DictPropertyInstance, v: T) -> None:
-        assert isinstance(instance, (EventBase, _EventInternalMetadata))
+        assert isinstance(instance, EventBase)
         instance._dict[self.key] = v
 
     def __delete__(self, instance: _DictPropertyInstance) -> None:
-        assert isinstance(instance, (EventBase, _EventInternalMetadata))
+        assert isinstance(instance, EventBase)
         try:
             del instance._dict[self.key]
         except KeyError as e1:
@@ -176,125 +177,10 @@ def __get__(
     ) -> Union[T, "DefaultDictProperty"]:
         if instance is None:
             return self
-        assert isinstance(instance, (EventBase, _EventInternalMetadata))
+        assert isinstance(instance, EventBase)
         return instance._dict.get(self.key, self.default)
 
 
-class _EventInternalMetadata:
-    __slots__ = ["_dict", "stream_ordering", "outlier"]
-
-    def __init__(self, internal_metadata_dict: JsonDict):
-        # we have to copy the dict, because it turns out that the same dict is
-        # reused. TODO: fix that
-        self._dict = dict(internal_metadata_dict)
-
-        # the stream ordering of this event. None, until it has been persisted.
-        self.stream_ordering: Optional[int] = None
-
-        # whether this event is an outlier (ie, whether we have the state at that point
-        # in the DAG)
-        self.outlier = False
-
-    out_of_band_membership: DictProperty[bool] = DictProperty("out_of_band_membership")
-    send_on_behalf_of: DictProperty[str] = DictProperty("send_on_behalf_of")
-    recheck_redaction: DictProperty[bool] = DictProperty("recheck_redaction")
-    soft_failed: DictProperty[bool] = DictProperty("soft_failed")
-    proactively_send: DictProperty[bool] = DictProperty("proactively_send")
-    redacted: DictProperty[bool] = DictProperty("redacted")
-
-    txn_id: DictProperty[str] = DictProperty("txn_id")
-    """The transaction ID, if it was set when the event was created."""
-
-    token_id: DictProperty[int] = DictProperty("token_id")
-    """The access token ID of the user who sent this event, if any."""
-
-    device_id: DictProperty[str] = DictProperty("device_id")
-    """The device ID of the user who sent this event, if any."""
-
-    # XXX: These are set by StreamWorkerStore._set_before_and_after.
-    # I'm pretty sure that these are never persisted to the database, so shouldn't
-    # be here
-    before: DictProperty[RoomStreamToken] = DictProperty("before")
-    after: DictProperty[RoomStreamToken] = DictProperty("after")
-    order: DictProperty[Tuple[int, int]] = DictProperty("order")
-
-    def get_dict(self) -> JsonDict:
-        return dict(self._dict)
-
-    def is_outlier(self) -> bool:
-        return self.outlier
-
-    def is_out_of_band_membership(self) -> bool:
-        """Whether this event is an out-of-band membership.
-
-        OOB memberships are a special case of outlier events: they are membership events
-        for federated rooms that we aren't full members of. Examples include invites
-        received over federation, and rejections for such invites.
-
-        The concept of an OOB membership is needed because these events need to be
-        processed as if they're new regular events (e.g. updating membership state in
-        the database, relaying to clients via /sync, etc) despite being outliers.
-
-        See also https://element-hq.github.io/synapse/develop/development/room-dag-concepts.html#out-of-band-membership-events.
-
-        (Added in synapse 0.99.0, so may be unreliable for events received before that)
-        """
-        return self._dict.get("out_of_band_membership", False)
-
-    def get_send_on_behalf_of(self) -> Optional[str]:
-        """Whether this server should send the event on behalf of another server.
-        This is used by the federation "send_join" API to forward the initial join
-        event for a server in the room.
-
-        returns a str with the name of the server this event is sent on behalf of.
-        """
-        return self._dict.get("send_on_behalf_of")
-
-    def need_to_check_redaction(self) -> bool:
-        """Whether the redaction event needs to be rechecked when fetching
-        from the database.
-
-        Starting in room v3 redaction events are accepted up front, and later
-        checked to see if the redacter and redactee's domains match.
-
-        If the sender of the redaction event is allowed to redact any event
-        due to auth rules, then this will always return false.
-        """
-        return self._dict.get("recheck_redaction", False)
-
-    def is_soft_failed(self) -> bool:
-        """Whether the event has been soft failed.
-
-        Soft failed events should be handled as usual, except:
-            1. They should not go down sync or event streams, or generally
-               sent to clients.
-            2. They should not be added to the forward extremities (and
-               therefore not to current state).
-        """
-        return self._dict.get("soft_failed", False)
-
-    def should_proactively_send(self) -> bool:
-        """Whether the event, if ours, should be sent to other clients and
-        servers.
-
-        This is used for sending dummy events internally. Servers and clients
-        can still explicitly fetch the event.
-        """
-        return self._dict.get("proactively_send", True)
-
-    def is_redacted(self) -> bool:
-        """Whether the event has been redacted.
-
-        This is used for efficiently checking whether an event has been
-        marked as redacted without needing to make another database call.
-        """
-        return self._dict.get("redacted", False)
-
-    def is_notifiable(self) -> bool:
-        """Whether this event can trigger a push notification"""
-        return not self.is_outlier() or self.is_out_of_band_membership()
-
-
 class EventBase(metaclass=abc.ABCMeta):
     @property
     @abc.abstractmethod
@@ -320,7 +206,7 @@ def __init__(
 
         self._dict = event_dict
 
-        self.internal_metadata = _EventInternalMetadata(internal_metadata_dict)
+        self.internal_metadata = EventInternalMetadata(internal_metadata_dict)
 
     depth: DictProperty[int] = DictProperty("depth")
     content: DictProperty[JsonDict] = DictProperty("content")
diff --git a/synapse/events/builder.py b/synapse/events/builder.py
index ae7092daaad..f32449c7da9 100644
--- a/synapse/events/builder.py
+++ b/synapse/events/builder.py
@@ -31,9 +31,10 @@
 )
 from synapse.crypto.event_signing import add_hashes_and_signatures
 from synapse.event_auth import auth_types_for_event
-from synapse.events import EventBase, _EventInternalMetadata, make_event_from_dict
+from synapse.events import EventBase, make_event_from_dict
 from synapse.state import StateHandler
 from synapse.storage.databases.main import DataStore
+from synapse.synapse_rust.events import EventInternalMetadata
 from synapse.types import EventID, JsonDict, StrCollection
 from synapse.types.state import StateFilter
 from synapse.util import Clock
@@ -93,8 +94,8 @@ class EventBuilder:
     _redacts: Optional[str] = None
     _origin_server_ts: Optional[int] = None
 
-    internal_metadata: _EventInternalMetadata = attr.Factory(
-        lambda: _EventInternalMetadata({})
+    internal_metadata: EventInternalMetadata = attr.Factory(
+        lambda: EventInternalMetadata({})
     )
 
     @property
diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py
index c4120630910..e3679d8f373 100644
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -1155,7 +1155,7 @@ async def _execute(pdu: EventBase) -> None:
             # NB: We *need* to copy to ensure that we don't have multiple
             # references being passed on, as that causes... issues.
             for s in signed_state:
-                s.internal_metadata = copy.deepcopy(s.internal_metadata)
+                s.internal_metadata = s.internal_metadata.copy()
 
             # double-check that the auth chain doesn't include a different create event
             auth_chain_create_events = [
diff --git a/synapse/federation/transport/server/__init__.py b/synapse/federation/transport/server/__init__.py
index 839092e4d35..74391e3cb2c 100644
--- a/synapse/federation/transport/server/__init__.py
+++ b/synapse/federation/transport/server/__init__.py
@@ -154,7 +154,10 @@ async def on_GET(
             limit = None
 
         data = await self.handler.get_local_public_room_list(
-            limit, since_token, network_tuple=network_tuple, from_federation=True
+            limit,
+            since_token,
+            network_tuple=network_tuple,
+            from_federation_origin=origin,
         )
         return 200, data
 
@@ -195,7 +198,7 @@ async def on_POST(
             since_token=since_token,
             search_filter=search_filter,
             network_tuple=network_tuple,
-            from_federation=True,
+            from_federation_origin=origin,
         )
 
         return 200, data
diff --git a/synapse/handlers/admin.py b/synapse/handlers/admin.py
index 9a4af3c45fe..db80345b945 100644
--- a/synapse/handlers/admin.py
+++ b/synapse/handlers/admin.py
@@ -208,7 +208,12 @@ async def export_user_data(self, user_id: str, writer: "ExfiltrationWriter") ->
                 if not events:
                     break
 
-                from_key = events[-1].internal_metadata.after
+                last_event = events[-1]
+                assert last_event.internal_metadata.stream_ordering
+                from_key = RoomStreamToken(
+                    stream=last_event.internal_metadata.stream_ordering,
+                    topological=last_event.depth,
+                )
 
                 events = await filter_events_for_client(
                     self._storage_controllers, user_id, events
diff --git a/synapse/handlers/federation_event.py b/synapse/handlers/federation_event.py
index 882be905dbb..12837429b9f 100644
--- a/synapse/handlers/federation_event.py
+++ b/synapse/handlers/federation_event.py
@@ -94,7 +94,7 @@
 )
 from synapse.types.state import StateFilter
 from synapse.util.async_helpers import Linearizer, concurrently_execute
-from synapse.util.iterutils import batch_iter, partition, sorted_topologically_batched
+from synapse.util.iterutils import batch_iter, partition, sorted_topologically
 from synapse.util.retryutils import NotRetryingDestination
 from synapse.util.stringutils import shortstr
 
@@ -1141,16 +1141,8 @@ async def _compute_event_context_with_maybe_missing_prevs(
             partial_state_flags = await self._store.get_partial_state_events(seen)
             partial_state = any(partial_state_flags.values())
 
-            # Get the state of the events we know about
-            ours = await self._state_storage_controller.get_state_groups_ids(
-                room_id, seen, await_full_state=False
-            )
-
             # state_maps is a list of mappings from (type, state_key) to event_id
-            state_maps: List[StateMap[str]] = list(ours.values())
-
-            # we don't need this any more, let's delete it.
-            del ours
+            state_maps: List[StateMap[str]] = []
 
             # Ask the remote server for the states we don't
             # know about
@@ -1169,6 +1161,17 @@ async def _compute_event_context_with_maybe_missing_prevs(
 
                     state_maps.append(remote_state_map)
 
+            # Get the state of the events we know about. We do this *after*
+            # trying to fetch missing state over federation as that might fail
+            # and then we can skip loading the local state.
+            ours = await self._state_storage_controller.get_state_groups_ids(
+                room_id, seen, await_full_state=False
+            )
+            state_maps.extend(ours.values())
+
+            # we don't need this any more, let's delete it.
+            del ours
+
             room_version = await self._store.get_room_version_id(room_id)
             state_map = await self._state_resolution_handler.resolve_events_with_store(
                 room_id,
@@ -1678,57 +1681,36 @@ async def _auth_and_persist_outliers(
 
         # We need to persist an event's auth events before the event.
         auth_graph = {
-            ev: [event_map[e_id] for e_id in ev.auth_event_ids() if e_id in event_map]
+            ev.event_id: [e_id for e_id in ev.auth_event_ids() if e_id in event_map]
             for ev in event_map.values()
         }
-        for roots in sorted_topologically_batched(event_map.values(), auth_graph):
-            if not roots:
-                # if *none* of the remaining events are ready, that means
-                # we have a loop. This either means a bug in our logic, or that
-                # somebody has managed to create a loop (which requires finding a
-                # hash collision in room v2 and later).
-                logger.warning(
-                    "Loop found in auth events while fetching missing state/auth "
-                    "events: %s",
-                    shortstr(event_map.keys()),
-                )
-                return
-
-            logger.info(
-                "Persisting %i of %i remaining outliers: %s",
-                len(roots),
-                len(event_map),
-                shortstr(e.event_id for e in roots),
-            )
-
-            await self._auth_and_persist_outliers_inner(room_id, roots)
-
-    async def _auth_and_persist_outliers_inner(
-        self, room_id: str, fetched_events: Collection[EventBase]
-    ) -> None:
-        """Helper for _auth_and_persist_outliers
-
-        Persists a batch of events where we have (theoretically) already persisted all
-        of their auth events.
-
-        Marks the events as outliers, auths them, persists them to the database, and,
-        where appropriate (eg, an invite), awakes the notifier.
+        sorted_auth_event_ids = sorted_topologically(event_map.keys(), auth_graph)
+        sorted_auth_events = [event_map[e_id] for e_id in sorted_auth_event_ids]
+        logger.info(
+            "Persisting %i remaining outliers: %s",
+            len(sorted_auth_events),
+            shortstr(e.event_id for e in sorted_auth_events),
+        )
 
-        Params:
-            origin: where the events came from
-            room_id: the room that the events are meant to be in (though this has
-               not yet been checked)
-            fetched_events: the events to persist
-        """
         # get all the auth events for all the events in this batch. By now, they should
         # have been persisted.
-        auth_events = {
-            aid for event in fetched_events for aid in event.auth_event_ids()
+        auth_event_ids = {
+            aid for event in sorted_auth_events for aid in event.auth_event_ids()
         }
-        persisted_events = await self._store.get_events(
-            auth_events,
-            allow_rejected=True,
-        )
+        auth_map = {
+            ev.event_id: ev
+            for ev in sorted_auth_events
+            if ev.event_id in auth_event_ids
+        }
+
+        missing_events = auth_event_ids.difference(auth_map)
+        if missing_events:
+            persisted_events = await self._store.get_events(
+                missing_events,
+                allow_rejected=True,
+                redact_behaviour=EventRedactBehaviour.as_is,
+            )
+            auth_map.update(persisted_events)
 
         events_and_contexts_to_persist: List[Tuple[EventBase, EventContext]] = []
 
@@ -1736,7 +1718,7 @@ async def prep(event: EventBase) -> None:
             with nested_logging_context(suffix=event.event_id):
                 auth = []
                 for auth_event_id in event.auth_event_ids():
-                    ae = persisted_events.get(auth_event_id)
+                    ae = auth_map.get(auth_event_id)
                     if not ae:
                         # the fact we can't find the auth event doesn't mean it doesn't
                         # exist, which means it is premature to reject `event`. Instead we
@@ -1755,7 +1737,9 @@ async def prep(event: EventBase) -> None:
                 context = EventContext.for_outlier(self._storage_controllers)
                 try:
                     validate_event_for_room_version(event)
-                    await check_state_independent_auth_rules(self._store, event)
+                    await check_state_independent_auth_rules(
+                        self._store, event, batched_auth_events=auth_map
+                    )
                     check_state_dependent_auth_rules(event, auth)
                 except AuthError as e:
                     logger.warning("Rejecting %r because %s", event, e)
@@ -1772,7 +1756,7 @@ async def prep(event: EventBase) -> None:
 
             events_and_contexts_to_persist.append((event, context))
 
-        for event in fetched_events:
+        for event in sorted_auth_events:
             await prep(event)
 
         await self.persist_events_and_notify(
diff --git a/synapse/handlers/relations.py b/synapse/handlers/relations.py
index 444b795872f..642f7acf97b 100644
--- a/synapse/handlers/relations.py
+++ b/synapse/handlers/relations.py
@@ -180,6 +180,10 @@ async def get_relations(
                 config=serialize_options,
             ),
         }
+
+        if recurse:
+            return_value["recursion_depth"] = 3
+
         if include_original_event:
             # Do not bundle aggregations when retrieving the original event because
             # we want the content before relations are applied to it.
diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index e78e598d5eb..b49b917b6ef 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -26,7 +26,17 @@
 import string
 from collections import OrderedDict
 from http import HTTPStatus
-from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, List, Optional, Tuple
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Awaitable,
+    Callable,
+    Dict,
+    List,
+    Optional,
+    Tuple,
+    cast,
+)
 
 import attr
 from typing_extensions import TypedDict
@@ -1742,13 +1752,19 @@ async def get_new_events(
             events = list(room_events)
             events.extend(e for evs, _ in room_to_events.values() for e in evs)
 
-            events.sort(key=lambda e: e.internal_metadata.order)
+            # We know stream_ordering must be not None here, as its been
+            # persisted, but mypy doesn't know that
+            events.sort(key=lambda e: cast(int, e.internal_metadata.stream_ordering))
 
             if limit:
                 events[:] = events[:limit]
 
             if events:
-                end_key = events[-1].internal_metadata.after
+                last_event = events[-1]
+                assert last_event.internal_metadata.stream_ordering
+                end_key = RoomStreamToken(
+                    stream=last_event.internal_metadata.stream_ordering,
+                )
             else:
                 end_key = to_key
 
diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py
index bfd205db494..5bad133f6b5 100644
--- a/synapse/handlers/room_list.py
+++ b/synapse/handlers/room_list.py
@@ -19,7 +19,7 @@
 #
 
 import logging
-from typing import TYPE_CHECKING, Any, Optional, Tuple
+from typing import TYPE_CHECKING, Any, List, Optional, Tuple
 
 import attr
 import msgpack
@@ -54,6 +54,9 @@
 # This is used to indicate we should only return rooms published to the main list.
 EMPTY_THIRD_PARTY_ID = ThirdPartyInstanceID(None, None)
 
+# Maximum number of local public rooms returned over the CS or SS API
+MAX_PUBLIC_ROOMS_IN_RESPONSE = 100
+
 
 class RoomListHandler:
     def __init__(self, hs: "HomeServer"):
@@ -74,7 +77,7 @@ async def get_local_public_room_list(
         since_token: Optional[str] = None,
         search_filter: Optional[dict] = None,
         network_tuple: Optional[ThirdPartyInstanceID] = EMPTY_THIRD_PARTY_ID,
-        from_federation: bool = False,
+        from_federation_origin: Optional[str] = None,
     ) -> JsonDict:
         """Generate a local public room list.
 
@@ -89,7 +92,8 @@ async def get_local_public_room_list(
                 This can be (None, None) to indicate the main list, or a particular
                 appservice and network id to use an appservice specific one.
                 Setting to None returns all public rooms across all lists.
-            from_federation: true iff the request comes from the federation API
+            from_federation_origin: the server name of the requester, or None
+                if the request is not from federation.
         """
         if not self.enable_room_list_search:
             return {"chunk": [], "total_room_count_estimate": 0}
@@ -102,36 +106,43 @@ async def get_local_public_room_list(
             network_tuple,
         )
 
-        if search_filter:
+        capped_limit: int = (
+            MAX_PUBLIC_ROOMS_IN_RESPONSE
+            if limit is None or limit > MAX_PUBLIC_ROOMS_IN_RESPONSE
+            else limit
+        )
+
+        if search_filter or from_federation_origin is not None:
             # We explicitly don't bother caching searches or requests for
             # appservice specific lists.
-            logger.info("Bypassing cache as search request.")
+            # We also don't bother caching requests from federated homeservers.
+            logger.debug("Bypassing cache as search or federation request.")
 
             return await self._get_public_room_list(
-                limit,
+                capped_limit,
                 since_token,
                 search_filter,
                 network_tuple=network_tuple,
-                from_federation=from_federation,
+                from_federation_origin=from_federation_origin,
             )
 
-        key = (limit, since_token, network_tuple)
+        key = (capped_limit, since_token, network_tuple)
         return await self.response_cache.wrap(
             key,
             self._get_public_room_list,
-            limit,
+            capped_limit,
             since_token,
             network_tuple=network_tuple,
-            from_federation=from_federation,
+            from_federation_origin=from_federation_origin,
         )
 
     async def _get_public_room_list(
         self,
-        limit: Optional[int] = None,
+        limit: int,
         since_token: Optional[str] = None,
         search_filter: Optional[dict] = None,
         network_tuple: Optional[ThirdPartyInstanceID] = EMPTY_THIRD_PARTY_ID,
-        from_federation: bool = False,
+        from_federation_origin: Optional[str] = None,
     ) -> JsonDict:
         """Generate a public room list.
         Args:
@@ -142,8 +153,8 @@ async def _get_public_room_list(
                 This can be (None, None) to indicate the main list, or a particular
                 appservice and network id to use an appservice specific one.
                 Setting to None returns all public rooms across all lists.
-            from_federation: Whether this request originated from a
-                federating server or a client. Used for room filtering.
+            from_federation_origin: the server name of the requester, or None
+                if the request is not from federation.
         """
 
         # Pagination tokens work by storing the room ID sent in the last batch,
@@ -165,8 +176,16 @@ async def _get_public_room_list(
             forwards = True
             has_batch_token = False
 
-        # we request one more than wanted to see if there are more pages to come
-        probing_limit = limit + 1 if limit is not None else None
+        if from_federation_origin is None:
+            # Client-Server API:
+            # we request one more than wanted to see if there are more pages to come
+            probing_limit = limit + 1
+        else:
+            # Federation API:
+            # we request a handful more in case any get filtered out by ACLs
+            # as a best easy effort attempt to return the full number of entries
+            # specified by `limit`.
+            probing_limit = limit + 10
 
         results = await self.store.get_largest_public_rooms(
             network_tuple,
@@ -174,7 +193,7 @@ async def _get_public_room_list(
             probing_limit,
             bounds=bounds,
             forwards=forwards,
-            ignore_non_federatable=from_federation,
+            ignore_non_federatable=from_federation_origin is not None,
         )
 
         def build_room_entry(room: LargestRoomStats) -> JsonDict:
@@ -195,59 +214,117 @@ def build_room_entry(room: LargestRoomStats) -> JsonDict:
             # Filter out Nones – rather omit the field altogether
             return {k: v for k, v in entry.items() if v is not None}
 
-        response: JsonDict = {}
+        # Build a list of up to `limit` entries.
+        room_entries: List[JsonDict] = []
+        rooms_iterator = results if forwards else reversed(results)
+
+        # Track the first and last 'considered' rooms so that we can provide correct
+        # next_batch/prev_batch tokens.
+        # This is needed because the loop might finish early when
+        # `len(room_entries) >= limit` and we might be left with rooms we didn't
+        # 'consider' (iterate over) and we should save those rooms for the next
+        # batch.
+        first_considered_room: Optional[LargestRoomStats] = None
+        last_considered_room: Optional[LargestRoomStats] = None
+        cut_off_due_to_limit: bool = False
+
+        for room_result in rooms_iterator:
+            if len(room_entries) >= limit:
+                cut_off_due_to_limit = True
+                break
+
+            if first_considered_room is None:
+                first_considered_room = room_result
+            last_considered_room = room_result
+
+            if from_federation_origin is not None:
+                # If this is a federated request, apply server ACLs if the room has any set
+                acl_evaluator = (
+                    await self._storage_controllers.state.get_server_acl_for_room(
+                        room_result.room_id
+                    )
+                )
+
+                if acl_evaluator is not None:
+                    if not acl_evaluator.server_matches_acl_event(
+                        from_federation_origin
+                    ):
+                        # the requesting server is ACL blocked by the room,
+                        # don't show in directory
+                        continue
+
+            room_entries.append(build_room_entry(room_result))
+
+        if not forwards:
+            # If we are paginating backwards, we still return the chunk in
+            # biggest-first order, so reverse again.
+            room_entries.reverse()
+            # Swap the order of first/last considered rooms.
+            first_considered_room, last_considered_room = (
+                last_considered_room,
+                first_considered_room,
+            )
+
+        response: JsonDict = {
+            "chunk": room_entries,
+        }
         num_results = len(results)
-        if limit is not None:
-            more_to_come = num_results == probing_limit
 
-            # Depending on direction we trim either the front or back.
-            if forwards:
-                results = results[:limit]
+        more_to_come_from_database = num_results == probing_limit
+
+        if forwards and has_batch_token:
+            # If there was a token given then we assume that there
+            # must be previous results, even if there were no results in this batch.
+            if first_considered_room is not None:
+                response["prev_batch"] = RoomListNextBatch(
+                    last_joined_members=first_considered_room.joined_members,
+                    last_room_id=first_considered_room.room_id,
+                    direction_is_forward=False,
+                ).to_token()
             else:
-                results = results[-limit:]
-        else:
-            more_to_come = False
+                # If we didn't find any results this time,
+                # we don't have an actual room ID to put in the token.
+                # But since `first_considered_room` is None, we know that we have
+                # reached the end of the results.
+                # So we can use a token of (0, empty room ID) to paginate from the end
+                # next time.
+                response["prev_batch"] = RoomListNextBatch(
+                    last_joined_members=0,
+                    last_room_id="",
+                    direction_is_forward=False,
+                ).to_token()
 
         if num_results > 0:
-            final_entry = results[-1]
-            initial_entry = results[0]
-
+            assert first_considered_room is not None
+            assert last_considered_room is not None
             if forwards:
-                if has_batch_token:
-                    # If there was a token given then we assume that there
-                    # must be previous results.
-                    response["prev_batch"] = RoomListNextBatch(
-                        last_joined_members=initial_entry.joined_members,
-                        last_room_id=initial_entry.room_id,
-                        direction_is_forward=False,
-                    ).to_token()
-
-                if more_to_come:
+                if more_to_come_from_database or cut_off_due_to_limit:
                     response["next_batch"] = RoomListNextBatch(
-                        last_joined_members=final_entry.joined_members,
-                        last_room_id=final_entry.room_id,
+                        last_joined_members=last_considered_room.joined_members,
+                        last_room_id=last_considered_room.room_id,
                         direction_is_forward=True,
                     ).to_token()
-            else:
+            else:  # backwards
                 if has_batch_token:
                     response["next_batch"] = RoomListNextBatch(
-                        last_joined_members=final_entry.joined_members,
-                        last_room_id=final_entry.room_id,
+                        last_joined_members=last_considered_room.joined_members,
+                        last_room_id=last_considered_room.room_id,
                         direction_is_forward=True,
                     ).to_token()
 
-                if more_to_come:
+                if more_to_come_from_database or cut_off_due_to_limit:
                     response["prev_batch"] = RoomListNextBatch(
-                        last_joined_members=initial_entry.joined_members,
-                        last_room_id=initial_entry.room_id,
+                        last_joined_members=first_considered_room.joined_members,
+                        last_room_id=first_considered_room.room_id,
                         direction_is_forward=False,
                     ).to_token()
 
-        response["chunk"] = [build_room_entry(r) for r in results]
-
+        # We can't efficiently count the total number of rooms that are not
+        # blocked by ACLs, but this is just an estimate so that should be
+        # good enough.
         response["total_room_count_estimate"] = await self.store.count_public_rooms(
             network_tuple,
-            ignore_non_federatable=from_federation,
+            ignore_non_federatable=from_federation_origin is not None,
             search_filter=search_filter,
         )
 
diff --git a/synapse/handlers/room_summary.py b/synapse/handlers/room_summary.py
index a534f5f280b..78bcac1429e 100644
--- a/synapse/handlers/room_summary.py
+++ b/synapse/handlers/room_summary.py
@@ -44,6 +44,7 @@
 from synapse.config.ratelimiting import RatelimitSettings
 from synapse.events import EventBase
 from synapse.types import JsonDict, Requester, StrCollection
+from synapse.types.state import StateFilter
 from synapse.util.caches.response_cache import ResponseCache
 
 if TYPE_CHECKING:
@@ -546,7 +547,16 @@ async def _is_local_room_accessible(
         Returns:
              True if the room is accessible to the requesting user or server.
         """
-        state_ids = await self._storage_controllers.state.get_current_state_ids(room_id)
+        event_types = [
+            (EventTypes.JoinRules, ""),
+            (EventTypes.RoomHistoryVisibility, ""),
+        ]
+        if requester:
+            event_types.append((EventTypes.Member, requester))
+
+        state_ids = await self._storage_controllers.state.get_current_state_ids(
+            room_id, state_filter=StateFilter.from_types(event_types)
+        )
 
         # If there's no state for the room, it isn't known.
         if not state_ids:
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 75250540c50..2e10035772f 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -602,7 +602,10 @@ async def _load_filtered_recents(
             if not limited or block_all_timeline:
                 prev_batch_token = upto_token
                 if recents:
-                    room_key = recents[0].internal_metadata.before
+                    assert recents[0].internal_metadata.stream_ordering
+                    room_key = RoomStreamToken(
+                        stream=recents[0].internal_metadata.stream_ordering - 1
+                    )
                     prev_batch_token = upto_token.copy_and_replace(
                         StreamKeyType.ROOM, room_key
                     )
@@ -691,7 +694,10 @@ async def _load_filtered_recents(
             if len(recents) > timeline_limit:
                 limited = True
                 recents = recents[-timeline_limit:]
-                room_key = recents[0].internal_metadata.before
+                assert recents[0].internal_metadata.stream_ordering
+                room_key = RoomStreamToken(
+                    stream=recents[0].internal_metadata.stream_ordering - 1
+                )
 
             prev_batch_token = upto_token.copy_and_replace(StreamKeyType.ROOM, room_key)
 
diff --git a/synapse/http/server.py b/synapse/http/server.py
index d570bf5e62a..31311aeb2a2 100644
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -574,7 +574,7 @@ def _send_response(
         assert isinstance(response_object, bytes)
         html_bytes = response_object
 
-        respond_with_html_bytes(request, 200, html_bytes)
+        respond_with_html_bytes(request, code, html_bytes)
 
     def _send_error_response(
         self,
diff --git a/synapse/rest/admin/users.py b/synapse/rest/admin/users.py
index 4059039bab7..ee73f0233dc 100644
--- a/synapse/rest/admin/users.py
+++ b/synapse/rest/admin/users.py
@@ -107,7 +107,7 @@ async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
             )
 
         user_id = parse_string(request, "user_id")
-        name = parse_string(request, "name")
+        name = parse_string(request, "name", encoding="utf-8")
 
         guests = parse_boolean(request, "guests", default=True)
         if self._msc3861_enabled and guests:
@@ -412,15 +412,6 @@ async def on_PUT(
                         target_user.to_string(), False, requester, by_admin=True
                     )
                 elif not deactivate and user["deactivated"]:
-                    if (
-                        "password" not in body
-                        and self.auth_handler.can_change_password()
-                    ):
-                        raise SynapseError(
-                            HTTPStatus.BAD_REQUEST,
-                            "Must provide a password to re-activate an account.",
-                        )
-
                     await self.deactivate_account_handler.activate_account(
                         target_user.to_string()
                     )
diff --git a/synapse/rest/synapse/client/password_reset.py b/synapse/rest/synapse/client/password_reset.py
index f6bdccad30c..671eae7c075 100644
--- a/synapse/rest/synapse/client/password_reset.py
+++ b/synapse/rest/synapse/client/password_reset.py
@@ -106,7 +106,7 @@ async def _async_render_POST(self, request: Request) -> Tuple[int, bytes]:
                     return (
                         302,
                         (
-                            b'You are being redirected to <a src="%s">%s</a>.'
+                            b'You are being redirected to <a href="%s">%s</a>.'
                             % (next_link_bytes, next_link_bytes)
                         ),
                     )
diff --git a/synapse/storage/databases/main/devices.py b/synapse/storage/databases/main/devices.py
index 66fabf4f764..ae914298fbc 100644
--- a/synapse/storage/databases/main/devices.py
+++ b/synapse/storage/databases/main/devices.py
@@ -1794,7 +1794,7 @@ async def delete_devices(self, user_id: str, device_ids: List[str]) -> None:
             device_ids: The IDs of the devices to delete
         """
 
-        def _delete_devices_txn(txn: LoggingTransaction) -> None:
+        def _delete_devices_txn(txn: LoggingTransaction, device_ids: List[str]) -> None:
             self.db_pool.simple_delete_many_txn(
                 txn,
                 table="devices",
@@ -1811,7 +1811,11 @@ def _delete_devices_txn(txn: LoggingTransaction) -> None:
                 keyvalues={"user_id": user_id},
             )
 
-        await self.db_pool.runInteraction("delete_devices", _delete_devices_txn)
+        for batch in batch_iter(device_ids, 100):
+            await self.db_pool.runInteraction(
+                "delete_devices", _delete_devices_txn, batch
+            )
+
         for device_id in device_ids:
             self.device_id_exists_cache.invalidate((user_id, device_id))
 
diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py
index 78ffeeaa465..1fd458b5102 100644
--- a/synapse/storage/databases/main/events_worker.py
+++ b/synapse/storage/databases/main/events_worker.py
@@ -1496,7 +1496,7 @@ def _fetch_event_rows(
                     room_version_id=row[5],
                     rejected_reason=row[6],
                     redactions=[],
-                    outlier=row[7],
+                    outlier=bool(row[7]),  # This is an int in SQLite3
                 )
 
             # check for redactions
diff --git a/synapse/storage/databases/main/state.py b/synapse/storage/databases/main/state.py
index 2d214476fc3..8006046453f 100644
--- a/synapse/storage/databases/main/state.py
+++ b/synapse/storage/databases/main/state.py
@@ -31,7 +31,10 @@
     Optional,
     Set,
     Tuple,
+    TypeVar,
+    Union,
     cast,
+    overload,
 )
 
 import attr
@@ -53,7 +56,7 @@
 )
 from synapse.storage.databases.main.events_worker import EventsWorkerStore
 from synapse.storage.databases.main.roommember import RoomMemberWorkerStore
-from synapse.types import JsonDict, JsonMapping, StateMap, StrCollection
+from synapse.types import JsonDict, JsonMapping, StateKey, StateMap, StrCollection
 from synapse.types.state import StateFilter
 from synapse.util.caches import intern_string
 from synapse.util.caches.descriptors import cached, cachedList
@@ -65,6 +68,8 @@
 
 logger = logging.getLogger(__name__)
 
+_T = TypeVar("_T")
+
 
 MAX_STATE_DELTA_HOPS = 100
 
@@ -364,7 +369,8 @@ async def get_partial_filtered_current_state_ids(
         def _get_filtered_current_state_ids_txn(
             txn: LoggingTransaction,
         ) -> StateMap[str]:
-            results = {}
+            results = StateMapWrapper(state_filter=state_filter or StateFilter.all())
+
             sql = """
                 SELECT type, state_key, event_id FROM current_state_events
                 WHERE room_id = ?
@@ -741,3 +747,41 @@ def __init__(
         hs: "HomeServer",
     ):
         super().__init__(database, db_conn, hs)
+
+
+@attr.s(auto_attribs=True, slots=True)
+class StateMapWrapper(Dict[StateKey, str]):
+    """A wrapper around a StateMap[str] to ensure that we only query for items
+    that were not filtered out.
+
+    This is to help prevent bugs where we filter out state but other bits of the
+    code expect the state to be there.
+    """
+
+    state_filter: StateFilter
+
+    def __getitem__(self, key: StateKey) -> str:
+        if key not in self.state_filter:
+            raise Exception("State map was filtered and doesn't include: %s", key)
+        return super().__getitem__(key)
+
+    @overload
+    def get(self, key: Tuple[str, str]) -> Optional[str]:
+        ...
+
+    @overload
+    def get(self, key: Tuple[str, str], default: Union[str, _T]) -> Union[str, _T]:
+        ...
+
+    def get(
+        self, key: StateKey, default: Union[str, _T, None] = None
+    ) -> Union[str, _T, None]:
+        if key not in self.state_filter:
+            raise Exception("State map was filtered and doesn't include: %s", key)
+        return super().get(key, default)
+
+    def __contains__(self, key: Any) -> bool:
+        if key not in self.state_filter:
+            raise Exception("State map was filtered and doesn't include: %s", key)
+
+        return super().__contains__(key)
diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py
index 1b2a65bed2b..aeeb74b46d5 100644
--- a/synapse/storage/databases/main/stream.py
+++ b/synapse/storage/databases/main/stream.py
@@ -705,8 +705,6 @@ def f(txn: LoggingTransaction) -> List[_EventDictReturn]:
             [r.event_id for r in rows], get_prev_content=True
         )
 
-        self._set_before_and_after(ret, rows, topo_order=False)
-
         if order.lower() == "desc":
             ret.reverse()
 
@@ -793,8 +791,6 @@ def f(txn: LoggingTransaction) -> List[_EventDictReturn]:
             [r.event_id for r in rows], get_prev_content=True
         )
 
-        self._set_before_and_after(ret, rows, topo_order=False)
-
         return ret
 
     async def get_recent_events_for_room(
@@ -820,8 +816,6 @@ async def get_recent_events_for_room(
             [r.event_id for r in rows], get_prev_content=True
         )
 
-        self._set_before_and_after(events, rows)
-
         return events, token
 
     async def get_recent_event_ids_for_room(
@@ -1094,31 +1088,6 @@ def _get_max_topological_txn(self, txn: LoggingTransaction, room_id: str) -> int
         # `[(None,)]`
         return rows[0][0] if rows[0][0] is not None else 0
 
-    @staticmethod
-    def _set_before_and_after(
-        events: List[EventBase], rows: List[_EventDictReturn], topo_order: bool = True
-    ) -> None:
-        """Inserts ordering information to events' internal metadata from
-        the DB rows.
-
-        Args:
-            events
-            rows
-            topo_order: Whether the events were ordered topologically or by stream
-                ordering. If true then all rows should have a non null
-                topological_ordering.
-        """
-        for event, row in zip(events, rows):
-            stream = row.stream_ordering
-            if topo_order and row.topological_ordering:
-                topo: Optional[int] = row.topological_ordering
-            else:
-                topo = None
-            internal = event.internal_metadata
-            internal.before = RoomStreamToken(topological=topo, stream=stream - 1)
-            internal.after = RoomStreamToken(topological=topo, stream=stream)
-            internal.order = (int(topo) if topo else 0, int(stream))
-
     async def get_events_around(
         self,
         room_id: str,
@@ -1559,8 +1528,6 @@ async def paginate_room_events(
             [r.event_id for r in rows], get_prev_content=True
         )
 
-        self._set_before_and_after(events, rows)
-
         return events, token
 
     @cached()
diff --git a/synapse/synapse_rust/events.pyi b/synapse/synapse_rust/events.pyi
new file mode 100644
index 00000000000..423ede59690
--- /dev/null
+++ b/synapse/synapse_rust/events.pyi
@@ -0,0 +1,106 @@
+# 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>.
+
+from typing import Optional
+
+from synapse.types import JsonDict
+
+class EventInternalMetadata:
+    def __init__(self, internal_metadata_dict: JsonDict): ...
+
+    stream_ordering: Optional[int]
+    """the stream ordering of this event. None, until it has been persisted."""
+
+    outlier: bool
+    """whether this event is an outlier (ie, whether we have the state at that
+    point in the DAG)"""
+
+    out_of_band_membership: bool
+    send_on_behalf_of: str
+    recheck_redaction: bool
+    soft_failed: bool
+    proactively_send: bool
+    redacted: bool
+
+    txn_id: str
+    """The transaction ID, if it was set when the event was created."""
+    token_id: int
+    """The access token ID of the user who sent this event, if any."""
+    device_id: str
+    """The device ID of the user who sent this event, if any."""
+
+    def get_dict(self) -> JsonDict: ...
+    def is_outlier(self) -> bool: ...
+    def copy(self) -> "EventInternalMetadata": ...
+    def is_out_of_band_membership(self) -> bool:
+        """Whether this event is an out-of-band membership.
+
+        OOB memberships are a special case of outlier events: they are membership events
+        for federated rooms that we aren't full members of. Examples include invites
+        received over federation, and rejections for such invites.
+
+        The concept of an OOB membership is needed because these events need to be
+        processed as if they're new regular events (e.g. updating membership state in
+        the database, relaying to clients via /sync, etc) despite being outliers.
+
+        See also https://element-hq.github.io/synapse/develop/development/room-dag-concepts.html#out-of-band-membership-events.
+
+        (Added in synapse 0.99.0, so may be unreliable for events received before that)
+        """
+        ...
+    def get_send_on_behalf_of(self) -> Optional[str]:
+        """Whether this server should send the event on behalf of another server.
+        This is used by the federation "send_join" API to forward the initial join
+        event for a server in the room.
+
+        returns a str with the name of the server this event is sent on behalf of.
+        """
+        ...
+    def need_to_check_redaction(self) -> bool:
+        """Whether the redaction event needs to be rechecked when fetching
+        from the database.
+
+        Starting in room v3 redaction events are accepted up front, and later
+        checked to see if the redacter and redactee's domains match.
+
+        If the sender of the redaction event is allowed to redact any event
+        due to auth rules, then this will always return false.
+        """
+        ...
+    def is_soft_failed(self) -> bool:
+        """Whether the event has been soft failed.
+
+        Soft failed events should be handled as usual, except:
+            1. They should not go down sync or event streams, or generally
+               sent to clients.
+            2. They should not be added to the forward extremities (and
+               therefore not to current state).
+        """
+        ...
+    def should_proactively_send(self) -> bool:
+        """Whether the event, if ours, should be sent to other clients and
+        servers.
+
+        This is used for sending dummy events internally. Servers and clients
+        can still explicitly fetch the event.
+        """
+        ...
+    def is_redacted(self) -> bool:
+        """Whether the event has been redacted.
+
+        This is used for efficiently checking whether an event has been
+        marked as redacted without needing to make another database call.
+        """
+        ...
+    def is_notifiable(self) -> bool:
+        """Whether this event can trigger a push notification"""
+        ...
diff --git a/synapse/types/state.py b/synapse/types/state.py
index 5ca3c94bceb..53662372af9 100644
--- a/synapse/types/state.py
+++ b/synapse/types/state.py
@@ -20,6 +20,7 @@
 import logging
 from typing import (
     TYPE_CHECKING,
+    Any,
     Callable,
     Collection,
     Dict,
@@ -584,6 +585,29 @@ def must_await_full_state(self, is_mine_id: Callable[[str], bool]) -> bool:
         # local users only
         return False
 
+    def __contains__(self, key: Any) -> bool:
+        if not isinstance(key, tuple) or len(key) != 2:
+            raise TypeError(
+                f"'in StateFilter' requires (str, str) as left operand, not {type(key).__name__}"
+            )
+
+        typ, state_key = key
+
+        if not isinstance(typ, str) or not isinstance(state_key, str):
+            raise TypeError(
+                f"'in StateFilter' requires (str, str) as left operand, not ({type(typ).__name__}, {type(state_key).__name__})"
+            )
+
+        if typ in self.types:
+            state_keys = self.types[typ]
+            if state_keys is None or state_key in state_keys:
+                return True
+
+        elif self.include_others:
+            return True
+
+        return False
+
 
 _ALL_STATE_FILTER = StateFilter(types=immutabledict(), include_others=True)
 _ALL_NON_MEMBER_STATE_FILTER = StateFilter(
diff --git a/tests/handlers/test_room_list.py b/tests/handlers/test_room_list.py
new file mode 100644
index 00000000000..4d22ef98c20
--- /dev/null
+++ b/tests/handlers/test_room_list.py
@@ -0,0 +1,88 @@
+from http import HTTPStatus
+from typing import Optional, Set
+
+from synapse.rest import admin
+from synapse.rest.client import directory, login, room
+from synapse.types import JsonDict
+
+from tests import unittest
+
+
+class RoomListHandlerTestCase(unittest.HomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+        directory.register_servlets,
+    ]
+
+    def _create_published_room(
+        self, tok: str, extra_content: Optional[JsonDict] = None
+    ) -> str:
+        room_id = self.helper.create_room_as(tok=tok, extra_content=extra_content)
+        channel = self.make_request(
+            "PUT",
+            f"/_matrix/client/v3/directory/list/room/{room_id}?access_token={tok}",
+            content={
+                "visibility": "public",
+            },
+        )
+        assert channel.code == HTTPStatus.OK, f"couldn't publish room: {channel.result}"
+        return room_id
+
+    def test_acls_applied_to_room_directory_results(self) -> None:
+        """
+        Creates 3 rooms. Room 2 has an ACL that only permits the homeservers
+        `test` and `test2` to access it.
+
+        We then simulate `test2` and `test3` requesting the room directory and assert
+        that `test3` does not see room 2, but `test2` sees all 3.
+        """
+        self.register_user("u1", "p1")
+        u1tok = self.login("u1", "p1")
+        room1 = self._create_published_room(u1tok)
+
+        room2 = self._create_published_room(
+            u1tok,
+            extra_content={
+                "initial_state": [
+                    {
+                        "type": "m.room.server_acl",
+                        "content": {
+                            "allow": ["test", "test2"],
+                        },
+                    }
+                ]
+            },
+        )
+
+        room3 = self._create_published_room(u1tok)
+
+        room_list = self.get_success(
+            self.hs.get_room_list_handler().get_local_public_room_list(
+                limit=50, from_federation_origin="test2"
+            )
+        )
+        room_ids_in_test2_list: Set[str] = {
+            entry["room_id"] for entry in room_list["chunk"]
+        }
+
+        room_list = self.get_success(
+            self.hs.get_room_list_handler().get_local_public_room_list(
+                limit=50, from_federation_origin="test3"
+            )
+        )
+        room_ids_in_test3_list: Set[str] = {
+            entry["room_id"] for entry in room_list["chunk"]
+        }
+
+        self.assertEqual(
+            room_ids_in_test2_list,
+            {room1, room2, room3},
+            "test2 should be able to see all 3 rooms",
+        )
+        self.assertEqual(
+            room_ids_in_test3_list,
+            {room1, room3},
+            "test3 should be able to see only 2 rooms",
+        )
diff --git a/tests/rest/admin/test_user.py b/tests/rest/admin/test_user.py
index 3caca98a35e..61cbac2332f 100644
--- a/tests/rest/admin/test_user.py
+++ b/tests/rest/admin/test_user.py
@@ -1638,8 +1638,17 @@ def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
             )
         )
 
+        self.non_ascii_displayname = "ąćęłńóśżźäöüß中国日本"
+        self.non_ascii_user = self.register_user(
+            "nonascii", "nonascii", displayname=self.non_ascii_displayname
+        )
+
         self.url_prefix = "/_synapse/admin/v2/users/%s"
         self.url_other_user = self.url_prefix % self.other_user
+        self.url_non_ascii_user = (
+            "/_synapse/admin/v2/users?name=%s"
+            % urllib.parse.quote(self.non_ascii_displayname)
+        )
 
     def test_requester_is_no_admin(self) -> None:
         """
@@ -1790,6 +1799,20 @@ def test_get_user(self) -> None:
         self.assertEqual("User", channel.json_body["displayname"])
         self._check_fields(channel.json_body)
 
+    def test_get_user_nonascii_displayname(self) -> None:
+        """
+        Test get user by non-ascii display name
+        """
+        channel = self.make_request(
+            "GET",
+            self.url_non_ascii_user,
+            access_token=self.admin_user_tok,
+        )
+
+        users = {user["name"]: user for user in channel.json_body["users"]}
+        self.assertEqual(200, channel.code, msg=channel.json_body)
+        self.assertIn(self.non_ascii_user, users, channel.json_body["users"])
+
     def test_create_server_admin(self) -> None:
         """
         Check that a new admin user is created successfully.
@@ -2747,7 +2770,7 @@ def test_change_name_deactivate_user_user_directory(self) -> None:
         profile = self.get_success(self.store._get_user_in_directory(self.other_user))
         self.assertIsNone(profile)
 
-    def test_reactivate_user(self) -> None:
+    def test_reactivate_user_with_password(self) -> None:
         """
         Test reactivating another user.
         """
@@ -2755,21 +2778,36 @@ def test_reactivate_user(self) -> None:
         # Deactivate the user.
         self._deactivate_user("@user:test")
 
-        # Attempt to reactivate the user (without a password).
+        # Reactivate the user with password.
         channel = self.make_request(
             "PUT",
             self.url_other_user,
             access_token=self.admin_user_tok,
-            content={"deactivated": False},
+            content={"deactivated": False, "password": "foo"},
         )
-        self.assertEqual(400, channel.code, msg=channel.json_body)
+        self.assertEqual(200, channel.code, msg=channel.json_body)
+        self.assertEqual("@user:test", channel.json_body["name"])
+        self.assertFalse(channel.json_body["deactivated"])
+        self._is_erased("@user:test", False)
+
+        # This key was removed intentionally. Ensure it is not accidentally re-included.
+        self.assertNotIn("password_hash", channel.json_body)
+
+    def test_reactivate_user_without_password(self) -> None:
+        """
+        Test reactivating another user without a password.
+        This can be using some local users and some user with SSO (password = `null`).
+        """
+
+        # Deactivate the user.
+        self._deactivate_user("@user:test")
 
-        # Reactivate the user.
+        # Reactivate the user without a password.
         channel = self.make_request(
             "PUT",
             self.url_other_user,
             access_token=self.admin_user_tok,
-            content={"deactivated": False, "password": "foo"},
+            content={"deactivated": False},
         )
         self.assertEqual(200, channel.code, msg=channel.json_body)
         self.assertEqual("@user:test", channel.json_body["name"])
@@ -2788,7 +2826,7 @@ def test_reactivate_user_localdb_disabled(self) -> None:
         # Deactivate the user.
         self._deactivate_user("@user:test")
 
-        # Reactivate the user with a password
+        # Reactivate the user with a password.
         channel = self.make_request(
             "PUT",
             self.url_other_user,
@@ -2822,7 +2860,7 @@ def test_reactivate_user_password_disabled(self) -> None:
         # Deactivate the user.
         self._deactivate_user("@user:test")
 
-        # Reactivate the user with a password
+        # Reactivate the user with a password.
         channel = self.make_request(
             "PUT",
             self.url_other_user,
diff --git a/tests/rest/client/test_account.py b/tests/rest/client/test_account.py
index 5cef9c5c17f..f1e4bdea76e 100644
--- a/tests/rest/client/test_account.py
+++ b/tests/rest/client/test_account.py
@@ -328,16 +328,49 @@ def test_password_reset_bad_email_inhibit_error(self) -> None:
 
         self.assertIsNotNone(session_id)
 
+    def test_password_reset_redirection(self) -> None:
+        """Test basic password reset flow"""
+        old_password = "monkey"
+
+        user_id = self.register_user("kermit", old_password)
+        self.login("kermit", old_password)
+
+        email = "test@example.com"
+
+        # Add a threepid
+        self.get_success(
+            self.store.user_add_threepid(
+                user_id=user_id,
+                medium="email",
+                address=email,
+                validated_at=0,
+                added_at=0,
+            )
+        )
+
+        client_secret = "foobar"
+        next_link = "http://example.com"
+        self._request_token(email, client_secret, "127.0.0.1", next_link)
+
+        self.assertEqual(len(self.email_attempts), 1)
+        link = self._get_link_from_email()
+
+        self._validate_token(link, next_link)
+
     def _request_token(
         self,
         email: str,
         client_secret: str,
         ip: str = "127.0.0.1",
+        next_link: Optional[str] = None,
     ) -> str:
+        body = {"client_secret": client_secret, "email": email, "send_attempt": 1}
+        if next_link is not None:
+            body["next_link"] = next_link
         channel = self.make_request(
             "POST",
             b"account/password/email/requestToken",
-            {"client_secret": client_secret, "email": email, "send_attempt": 1},
+            body,
             client_ip=ip,
         )
 
@@ -350,7 +383,7 @@ def _request_token(
 
         return channel.json_body["sid"]
 
-    def _validate_token(self, link: str) -> None:
+    def _validate_token(self, link: str, next_link: Optional[str] = None) -> None:
         # Remove the host
         path = link.replace("https://example.com", "")
 
@@ -378,7 +411,11 @@ def _validate_token(self, link: str) -> None:
             shorthand=False,
             content_is_form=True,
         )
-        self.assertEqual(HTTPStatus.OK, channel.code, channel.result)
+        self.assertEqual(
+            HTTPStatus.OK if next_link is None else HTTPStatus.FOUND,
+            channel.code,
+            channel.result,
+        )
 
     def _get_link_from_email(self) -> str:
         assert self.email_attempts, "No emails have been sent"
diff --git a/tests/storage/databases/main/test_events_worker.py b/tests/storage/databases/main/test_events_worker.py
index ed747a8b3ce..caa57520329 100644
--- a/tests/storage/databases/main/test_events_worker.py
+++ b/tests/storage/databases/main/test_events_worker.py
@@ -324,7 +324,7 @@ def _populate_events(self) -> None:
         )
 
         self.event_ids: List[str] = []
-        for idx in range(20):
+        for idx in range(1, 21):  # Stream ordering starts at 1.
             event_json = {
                 "type": f"test {idx}",
                 "room_id": self.room_id,
diff --git a/tests/storage/test_event_federation.py b/tests/storage/test_event_federation.py
index b3e0bc47ec9..0a6253e22c5 100644
--- a/tests/storage/test_event_federation.py
+++ b/tests/storage/test_event_federation.py
@@ -44,12 +44,13 @@
     EventFormatVersions,
     RoomVersion,
 )
-from synapse.events import EventBase, _EventInternalMetadata
+from synapse.events import EventBase
 from synapse.rest import admin
 from synapse.rest.client import login, room
 from synapse.server import HomeServer
 from synapse.storage.database import LoggingTransaction
 from synapse.storage.types import Cursor
+from synapse.synapse_rust.events import EventInternalMetadata
 from synapse.types import JsonDict
 from synapse.util import Clock, json_encoder
 
@@ -1209,7 +1210,7 @@ class FakeEvent:
     type = "foo"
     state_key = "foo"
 
-    internal_metadata = _EventInternalMetadata({})
+    internal_metadata = EventInternalMetadata({})
 
     def auth_event_ids(self) -> List[str]:
         return self.auth_events
diff --git a/tests/storage/test_redaction.py b/tests/storage/test_redaction.py
index c88ec69c94c..cb459d6b036 100644
--- a/tests/storage/test_redaction.py
+++ b/tests/storage/test_redaction.py
@@ -25,9 +25,10 @@
 
 from synapse.api.constants import EventTypes, Membership
 from synapse.api.room_versions import RoomVersions
-from synapse.events import EventBase, _EventInternalMetadata
+from synapse.events import EventBase
 from synapse.events.builder import EventBuilder
 from synapse.server import HomeServer
+from synapse.synapse_rust.events import EventInternalMetadata
 from synapse.types import JsonDict, RoomID, UserID
 from synapse.util import Clock
 
@@ -268,7 +269,7 @@ def type(self) -> str:
                 return self._base_builder.type
 
             @property
-            def internal_metadata(self) -> _EventInternalMetadata:
+            def internal_metadata(self) -> EventInternalMetadata:
                 return self._base_builder.internal_metadata
 
         event_1, unpersisted_context_1 = self.get_success(