Ran: nix flake update liburing

If the reverse_proxy directive is omitted before 127.0.0.1:6167 in your Caddyfile, enabling the service with systemctl enable will result in an error.

.gitattributes

@@ -0,0 +1,87 @@
+# taken from https://github.com/gitattributes/gitattributes/blob/46a8961ad73f5bd4d8d193708840fbc9e851d702/Rust.gitattributes
+# Auto detect text files and perform normalization
+*          text=auto
+
+*.rs       text diff=rust
+*.toml     text diff=toml
+Cargo.lock text
+
+# taken from https://github.com/gitattributes/gitattributes/blob/46a8961ad73f5bd4d8d193708840fbc9e851d702/Common.gitattributes
+# Documents
+*.bibtex   text diff=bibtex
+*.doc      diff=astextplain
+*.DOC      diff=astextplain
+*.docx     diff=astextplain
+*.DOCX     diff=astextplain
+*.dot      diff=astextplain
+*.DOT      diff=astextplain
+*.pdf      diff=astextplain
+*.PDF      diff=astextplain
+*.rtf      diff=astextplain
+*.RTF      diff=astextplain
+*.md       text diff=markdown
+*.mdx      text diff=markdown
+*.tex      text diff=tex
+*.adoc     text
+*.textile  text
+*.mustache text
+*.csv      text eol=crlf
+*.tab      text
+*.tsv      text
+*.txt      text
+*.sql      text
+*.epub     diff=astextplain
+
+# Graphics
+*.png      binary
+*.jpg      binary
+*.jpeg     binary
+*.gif      binary
+*.tif      binary
+*.tiff     binary
+*.ico      binary
+# SVG treated as text by default.
+*.svg      text
+*.eps      binary
+
+# Scripts
+*.bash     text eol=lf
+*.fish     text eol=lf
+*.ksh      text eol=lf
+*.sh       text eol=lf
+*.zsh      text eol=lf
+# These are explicitly windows files and should use crlf
+*.bat      text eol=crlf
+*.cmd      text eol=crlf
+*.ps1      text eol=crlf
+
+# Serialisation
+*.json     text
+*.toml     text
+*.xml      text
+*.yaml     text
+*.yml      text
+
+# Archives
+*.7z       binary
+*.bz       binary
+*.bz2      binary
+*.bzip2    binary
+*.gz       binary
+*.lz       binary
+*.lzma     binary
+*.rar      binary
+*.tar      binary
+*.taz      binary
+*.tbz      binary
+*.tbz2     binary
+*.tgz      binary
+*.tlz      binary
+*.txz      binary
+*.xz       binary
+*.Z        binary
+*.zip      binary
+*.zst      binary
+
+# Text files where line endings should be preserved
+*.patch    -text

.github/workflows/ci.yml

@@ -128,7 +128,7 @@ jobs:
             - name: Restore and cache Nix store
               # we want a fresh-state when we do releases/tags to avoid potential cache poisoning attacks impacting
               # releases and tags
-              if: ${{ !startsWith(github.ref, 'refs/tags/') }}
+              #if: ${{ !startsWith(github.ref, 'refs/tags/') }}
               uses: nix-community/cache-nix-action@v5.1.0
               with:
                 # restore and save a cache using this key
@@ -191,14 +191,14 @@ jobs:
             - name: Run sccache-cache
               # we want a fresh-state when we do releases/tags to avoid potential cache poisoning attacks impacting
               # releases and tags
-              if: ${{ (env.SCCACHE_GHA_ENABLED == 'true') && !startsWith(github.ref, 'refs/tags/') }}
+              #if: ${{ (env.SCCACHE_GHA_ENABLED == 'true') && !startsWith(github.ref, 'refs/tags/') }}
               uses: mozilla-actions/sccache-action@main
 
             # use rust-cache
             - uses: Swatinem/rust-cache@v2
               # we want a fresh-state when we do releases/tags to avoid potential cache poisoning attacks impacting
               # releases and tags
-              if: ${{ !startsWith(github.ref, 'refs/tags/') }}
+              #if: ${{ !startsWith(github.ref, 'refs/tags/') }}
               with:
                 cache-all-crates: "true"
                 cache-on-failure: "true"
@@ -323,7 +323,7 @@ jobs:
             - name: Restore and cache Nix store
               # we want a fresh-state when we do releases/tags to avoid potential cache poisoning attacks impacting
               # releases and tags
-              if: ${{ !startsWith(github.ref, 'refs/tags/') }}
+              #if: ${{ !startsWith(github.ref, 'refs/tags/') }}
               uses: nix-community/cache-nix-action@v5.1.0
               with:
                 # restore and save a cache using this key
@@ -379,14 +379,14 @@ jobs:
             - name: Run sccache-cache
               # we want a fresh-state when we do releases/tags to avoid potential cache poisoning attacks impacting
               # releases and tags
-              if: ${{ (env.SCCACHE_GHA_ENABLED == 'true') && !startsWith(github.ref, 'refs/tags/') }}
+              #if: ${{ (env.SCCACHE_GHA_ENABLED == 'true') && !startsWith(github.ref, 'refs/tags/') }}
               uses: mozilla-actions/sccache-action@main
 
             # use rust-cache
             - uses: Swatinem/rust-cache@v2
               # we want a fresh-state when we do releases/tags to avoid potential cache poisoning attacks impacting
               # releases and tags
-              if: ${{ !startsWith(github.ref, 'refs/tags/') }}
+              #if: ${{ !startsWith(github.ref, 'refs/tags/') }}
               with:
                 cache-all-crates: "true"
                 cache-on-failure: "true"
@@ -679,7 +679,7 @@ jobs:
             - name: Run sccache-cache
               # we want a fresh-state when we do releases/tags to avoid potential cache poisoning attacks impacting
               # releases and tags
-              if: ${{ (env.SCCACHE_GHA_ENABLED == 'true') && !startsWith(github.ref, 'refs/tags/') }}
+              #if: ${{ (env.SCCACHE_GHA_ENABLED == 'true') && !startsWith(github.ref, 'refs/tags/') }}
               uses: mozilla-actions/sccache-action@main
 
             # use rust-cache

Cargo.lock

@@ -26,6 +26,12 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "aligned-vec"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4aa90d7ce82d4be67b64039a3d588d38dbcc6736577de4a847025ce5b0c468d1"
+
 [[package]]
 name = "alloc-no-stdlib"
 version = "2.0.4"
@@ -53,12 +59,29 @@ version = "1.0.95"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "34ac096ce696dc2fcabef30516bb13c0a68a11d30131d3df6f04711467681b04"
 
+[[package]]
+name = "arbitrary"
+version = "1.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dde20b3d026af13f561bdd0f15edf01fc734f0dafcedbaf42bba506a9517f223"
+
 [[package]]
 name = "arc-swap"
 version = "1.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "69f7f8c3906b62b754cd5326047894316021dcfe5a194c8ea52bdd94934a3457"
 
+[[package]]
+name = "arg_enum_proc_macro"
+version = "0.3.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ae92a5119aa49cdbcf6b9f893fe4e1d98b04ccbf82ee0584ad948a44a734dea"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.96",
+]
+
 [[package]]
 name = "argon2"
 version = "0.5.3"
@@ -173,6 +196,29 @@ version = "1.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26"
 
+[[package]]
+name = "av1-grain"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6678909d8c5d46a42abcf571271e15fdbc0a225e3646cf23762cd415046c78bf"
+dependencies = [
+ "anyhow",
+ "arrayvec",
+ "log",
+ "nom",
+ "num-rational",
+ "v_frame",
+]
+
+[[package]]
+name = "avif-serialize"
+version = "0.8.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e335041290c43101ca215eed6f43ec437eb5a42125573f600fc3fa42b9bddd62"
+dependencies = [
+ "arrayvec",
+]
+
 [[package]]
 name = "aws-lc-rs"
 version = "1.12.1"
@@ -385,6 +431,12 @@ dependencies = [
  "which",
 ]
 
+[[package]]
+name = "bit_field"
+version = "0.10.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "dc827186963e592360843fb5ba4b973e145841266c1357f7180c43526f2e5b61"
+
 [[package]]
 name = "bitflags"
 version = "1.3.2"
@@ -397,6 +449,12 @@ version = "2.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8f68f53c83ab957f72c32642f3868eec03eb974d1fb82e453128456482613d36"
 
+[[package]]
+name = "bitstream-io"
+version = "2.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6099cdc01846bc367c4e7dd630dc5966dccf36b652fae7a74e17b640411a91b2"
+
 [[package]]
 name = "blake2"
 version = "0.10.6"
@@ -415,6 +473,15 @@ dependencies = [
  "generic-array",
 ]
 
+[[package]]
+name = "blurhash"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e79769241dcd44edf79a732545e8b5cec84c247ac060f5252cd51885d093a8fc"
+dependencies = [
+ "image",
+]
+
 [[package]]
 name = "brotli"
 version = "7.0.0"
@@ -436,6 +503,12 @@ dependencies = [
  "alloc-stdlib",
 ]
 
+[[package]]
+name = "built"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c360505aed52b7ec96a3636c3f039d99103c37d1d9b4f7a8c743d3ea9ffcd03b"
+
 [[package]]
 name = "bumpalo"
 version = "3.16.0"
@@ -513,6 +586,16 @@ dependencies = [
  "nom",
 ]
 
+[[package]]
+name = "cfg-expr"
+version = "0.15.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d067ad48b8650848b989a59a86c6c36a995d02d2bf778d45c3c5d57bc2718f02"
+dependencies = [
+ "smallvec",
+ "target-lexicon",
+]
+
 [[package]]
 name = "cfg-if"
 version = "1.0.0"
@@ -822,6 +905,7 @@ dependencies = [
  "arrayvec",
  "async-trait",
  "base64 0.22.1",
+ "blurhash",
  "bytes",
  "conduwuit_core",
  "conduwuit_database",
@@ -1071,6 +1155,12 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "crunchy"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43da5946c66ffcc7745f48db692ffbb10a83bfe0afd96235c5c2a4fb23994929"
+
 [[package]]
 name = "crypto-common"
 version = "0.1.6"
@@ -1252,7 +1342,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "33d852cb9b869c2a9b3df2f71a3074817f01e1844f839a144f5fcef059a4eb5d"
 dependencies = [
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -1275,6 +1365,21 @@ dependencies = [
  "pin-project-lite",
 ]
 
+[[package]]
+name = "exr"
+version = "1.73.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f83197f59927b46c04a183a619b7c29df34e63e63c7869320862268c0ef687e0"
+dependencies = [
+ "bit_field",
+ "half",
+ "lebe",
+ "miniz_oxide",
+ "rayon-core",
+ "smallvec",
+ "zune-inflate",
+]
+
 [[package]]
 name = "fdeflate"
 version = "0.3.7"
@@ -1519,6 +1624,16 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "half"
+version = "2.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6dd08c532ae367adf81c312a4580bc67f1d0fe8bc9c460520283f4c0ff277888"
+dependencies = [
+ "cfg-if",
+ "crunchy",
+]
+
 [[package]]
 name = "hardened_malloc-rs"
 version = "0.1.2+12"
@@ -1973,10 +2088,16 @@ dependencies = [
  "bytemuck",
  "byteorder-lite",
  "color_quant",
+ "exr",
  "gif",
  "image-webp",
  "num-traits",
  "png",
+ "qoi",
+ "ravif",
+ "rayon",
+ "rgb",
+ "tiff",
  "zune-core",
  "zune-jpeg",
 ]
@@ -1991,6 +2112,12 @@ dependencies = [
  "quick-error 2.0.1",
 ]
 
+[[package]]
+name = "imgref"
+version = "1.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d0263a3d970d5c054ed9312c0057b4f3bde9c0b33836d3637361d4a9e6e7a408"
+
 [[package]]
 name = "indexmap"
 version = "1.9.3"
@@ -2024,6 +2151,17 @@ version = "3.0.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02"
 
+[[package]]
+name = "interpolate_name"
+version = "0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c34819042dc3d3971c46c2190835914dfbe0c3c13f61449b2997f4e9722dfa60"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.96",
+]
+
 [[package]]
 name = "ipaddress"
 version = "0.1.3"
@@ -2089,6 +2227,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "jpeg-decoder"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f5d4a7da358eff58addd2877a45865158f0d78c911d43a5784ceb7bbf52833b0"
+
 [[package]]
 name = "js-sys"
 version = "0.3.77"
@@ -2172,12 +2316,28 @@ version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
 
+[[package]]
+name = "lebe"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "03087c2bad5e1034e8cace5926dec053fb3790248370865f5117a7d0213354c8"
+
 [[package]]
 name = "libc"
 version = "0.2.169"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b5aba8db14291edd000dfcc4d620c7ebfb122c613afb886ca8803fa4e128a20a"
 
+[[package]]
+name = "libfuzzer-sys"
+version = "0.4.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf78f52d400cf2d84a3a973a78a592b4adc535739e0a5597a0da6f0c357adc75"
+dependencies = [
+ "arbitrary",
+ "cc",
+]
+
 [[package]]
 name = "libloading"
 version = "0.8.6"
@@ -2185,7 +2345,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fc2f4eb4bc735547cfed7c0a4922cbd04a4655978c09b54f1f7b228750664c34"
 dependencies = [
  "cfg-if",
- "windows-targets 0.48.5",
+ "windows-targets 0.52.6",
 ]
 
 [[package]]
@@ -2243,6 +2403,15 @@ dependencies = [
  "futures-sink",
 ]
 
+[[package]]
+name = "loop9"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0fae87c125b03c1d2c0150c90365d7d6bcc53fb73a9acaef207d2d065860f062"
+dependencies = [
+ "imgref",
+]
+
 [[package]]
 name = "lru-cache"
 version = "0.1.2"
@@ -2321,6 +2490,16 @@ version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94"
 
+[[package]]
+name = "maybe-rayon"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8ea1f30cedd69f0a2954655f7188c6a834246d2bcf1e315e2ac40c4b24dc9519"
+dependencies = [
+ "cfg-if",
+ "rayon",
+]
+
 [[package]]
 name = "memchr"
 version = "2.7.4"
@@ -2434,6 +2613,12 @@ version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e9e591e719385e6ebaeb5ce5d3887f7d5676fceca6411d1925ccc95745f3d6f7"
 
+[[package]]
+name = "noop_proc_macro"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0676bb32a98c1a483ce53e500a81ad9c3d5b3f7c920c28c24e9cb0980d0b5bc8"
+
 [[package]]
 name = "nu-ansi-term"
 version = "0.46.0"
@@ -2483,6 +2668,17 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9"
 
+[[package]]
+name = "num-derive"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed3955f1a9c7c0c15e092f9c887db08b1fc683305fdf6eb6684f22555355e202"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.96",
+]
+
 [[package]]
 name = "num-integer"
 version = "0.1.46"
@@ -2907,6 +3103,25 @@ dependencies = [
  "yansi",
 ]
 
+[[package]]
+name = "profiling"
+version = "1.0.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "afbdc74edc00b6f6a218ca6a5364d6226a259d4b8ea1af4a0ea063f27e179f4d"
+dependencies = [
+ "profiling-procmacros",
+]
+
+[[package]]
+name = "profiling-procmacros"
+version = "1.0.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a65f2e60fbf1063868558d69c6beacf412dc755f9fc020f514b7955fc914fe30"
+dependencies = [
+ "quote",
+ "syn 2.0.96",
+]
+
 [[package]]
 name = "prost"
 version = "0.13.4"
@@ -2957,6 +3172,15 @@ version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "007d8adb5ddab6f8e3f491ac63566a7d5002cc7ed73901f72057943fa71ae1ae"
 
+[[package]]
+name = "qoi"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7f6d64c71eb498fe9eae14ce4ec935c555749aef511cca85b5568910d6e48001"
+dependencies = [
+ "bytemuck",
+]
+
 [[package]]
 name = "quick-error"
 version = "1.2.3"
@@ -3018,7 +3242,7 @@ dependencies = [
  "once_cell",
  "socket2",
  "tracing",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -3060,6 +3284,76 @@ dependencies = [
  "getrandom",
 ]
 
+[[package]]
+name = "rav1e"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd87ce80a7665b1cce111f8a16c1f3929f6547ce91ade6addf4ec86a8dda5ce9"
+dependencies = [
+ "arbitrary",
+ "arg_enum_proc_macro",
+ "arrayvec",
+ "av1-grain",
+ "bitstream-io",
+ "built",
+ "cfg-if",
+ "interpolate_name",
+ "itertools 0.12.1",
+ "libc",
+ "libfuzzer-sys",
+ "log",
+ "maybe-rayon",
+ "new_debug_unreachable",
+ "noop_proc_macro",
+ "num-derive",
+ "num-traits",
+ "once_cell",
+ "paste",
+ "profiling",
+ "rand",
+ "rand_chacha",
+ "simd_helpers",
+ "system-deps",
+ "thiserror 1.0.69",
+ "v_frame",
+ "wasm-bindgen",
+]
+
+[[package]]
+name = "ravif"
+version = "0.11.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2413fd96bd0ea5cdeeb37eaf446a22e6ed7b981d792828721e74ded1980a45c6"
+dependencies = [
+ "avif-serialize",
+ "imgref",
+ "loop9",
+ "quick-error 2.0.1",
+ "rav1e",
+ "rayon",
+ "rgb",
+]
+
+[[package]]
+name = "rayon"
+version = "1.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b418a60154510ca1a002a752ca9714984e21e4241e804d32555251faf8b78ffa"
+dependencies = [
+ "either",
+ "rayon-core",
+]
+
+[[package]]
+name = "rayon-core"
+version = "1.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1465873a3dfdaa8ae7cb14b4383657caab0b3e8a0aa9ae8e04b044854c8dfce2"
+dependencies = [
+ "crossbeam-deque",
+ "crossbeam-utils",
+]
+
 [[package]]
 name = "redox_syscall"
 version = "0.5.8"
@@ -3172,6 +3466,12 @@ dependencies = [
  "quick-error 1.2.3",
 ]
 
+[[package]]
+name = "rgb"
+version = "0.8.50"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57397d16646700483b67d2dd6511d79318f9d057fdbd21a4066aeac8b41d310a"
+
 [[package]]
 name = "ring"
 version = "0.17.8"
@@ -3190,7 +3490,7 @@ dependencies = [
 [[package]]
 name = "ruma"
 version = "0.10.1"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "assign",
  "js_int",
@@ -3212,7 +3512,7 @@ dependencies = [
 [[package]]
 name = "ruma-appservice-api"
 version = "0.10.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3224,7 +3524,7 @@ dependencies = [
 [[package]]
 name = "ruma-client-api"
 version = "0.18.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "as_variant",
  "assign",
@@ -3247,7 +3547,7 @@ dependencies = [
 [[package]]
 name = "ruma-common"
 version = "0.13.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "as_variant",
  "base64 0.22.1",
@@ -3278,7 +3578,7 @@ dependencies = [
 [[package]]
 name = "ruma-events"
 version = "0.28.1"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "as_variant",
  "indexmap 2.7.0",
@@ -3303,7 +3603,7 @@ dependencies = [
 [[package]]
 name = "ruma-federation-api"
 version = "0.9.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "bytes",
  "http",
@@ -3321,7 +3621,7 @@ dependencies = [
 [[package]]
 name = "ruma-identifiers-validation"
 version = "0.9.5"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "js_int",
  "thiserror 2.0.11",
@@ -3330,7 +3630,7 @@ dependencies = [
 [[package]]
 name = "ruma-identity-service-api"
 version = "0.9.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3340,7 +3640,7 @@ dependencies = [
 [[package]]
 name = "ruma-macros"
 version = "0.13.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "cfg-if",
  "proc-macro-crate",
@@ -3355,7 +3655,7 @@ dependencies = [
 [[package]]
 name = "ruma-push-gateway-api"
 version = "0.9.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "js_int",
  "ruma-common",
@@ -3367,7 +3667,7 @@ dependencies = [
 [[package]]
 name = "ruma-server-util"
 version = "0.3.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "headers",
  "http",
@@ -3380,7 +3680,7 @@ dependencies = [
 [[package]]
 name = "ruma-signatures"
 version = "0.15.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "base64 0.22.1",
  "ed25519-dalek",
@@ -3396,7 +3696,7 @@ dependencies = [
 [[package]]
 name = "ruma-state-res"
 version = "0.11.0"
-source = "git+https://github.com/girlbossceo/ruwuma?rev=b560338b2a50dbf61ecfe80808b9b095ad4cec00#b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+source = "git+https://github.com/girlbossceo/ruwuma?rev=f5667c6292adb43fbe4725d31d6b5127a0cf60ce#f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 dependencies = [
  "futures-util",
  "js_int",
@@ -3411,7 +3711,7 @@ dependencies = [
 [[package]]
 name = "rust-librocksdb-sys"
 version = "0.32.0+9.10.0"
-source = "git+https://github.com/girlbossceo/rust-rocksdb-zaidoon1?rev=1f032427d3a0e7b0f13c04b4e34712bd8610291b#1f032427d3a0e7b0f13c04b4e34712bd8610291b"
+source = "git+https://github.com/girlbossceo/rust-rocksdb-zaidoon1?rev=7b0e1bbe395a41ba8a11347a4921da590e3ad0d9#7b0e1bbe395a41ba8a11347a4921da590e3ad0d9"
 dependencies = [
  "bindgen",
  "bzip2-sys",
@@ -3428,7 +3728,7 @@ dependencies = [
 [[package]]
 name = "rust-rocksdb"
 version = "0.36.0"
-source = "git+https://github.com/girlbossceo/rust-rocksdb-zaidoon1?rev=1f032427d3a0e7b0f13c04b4e34712bd8610291b#1f032427d3a0e7b0f13c04b4e34712bd8610291b"
+source = "git+https://github.com/girlbossceo/rust-rocksdb-zaidoon1?rev=7b0e1bbe395a41ba8a11347a4921da590e3ad0d9#7b0e1bbe395a41ba8a11347a4921da590e3ad0d9"
 dependencies = [
  "libc",
  "rust-librocksdb-sys",
@@ -3479,7 +3779,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -3945,6 +4245,15 @@ version = "0.3.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d66dc143e6b11c1eddc06d5c423cfc97062865baf299914ab64caa38182078fe"
 
+[[package]]
+name = "simd_helpers"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "95890f873bec569a0362c235787f3aca6e1e887302ba4840839bcc6459c42da6"
+dependencies = [
+ "quote",
+]
+
 [[package]]
 name = "siphasher"
 version = "0.3.11"
@@ -4096,6 +4405,25 @@ dependencies = [
  "syn 2.0.96",
 ]
 
+[[package]]
+name = "system-deps"
+version = "6.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a3e535eb8dded36d55ec13eddacd30dec501792ff23a0b1682c38601b8cf2349"
+dependencies = [
+ "cfg-expr",
+ "heck",
+ "pkg-config",
+ "toml",
+ "version-compare",
+]
+
+[[package]]
+name = "target-lexicon"
+version = "0.12.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "61c41af27dd6d1e27b1b16b489db798443478cef1f06a660c96db617ba5de3b1"
+
 [[package]]
 name = "tendril"
 version = "0.4.3"
@@ -4205,6 +4533,17 @@ dependencies = [
  "threadpool",
 ]
 
+[[package]]
+name = "tiff"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba1310fcea54c6a9a4fd1aad794ecc02c31682f6bfbecdf460bf19533eed1e3e"
+dependencies = [
+ "flate2",
+ "jpeg-decoder",
+ "weezl",
+]
+
 [[package]]
 name = "tikv-jemalloc-ctl"
 version = "0.6.0"
@@ -4744,6 +5083,17 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "v_frame"
+version = "0.3.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d6f32aaa24bacd11e488aa9ba66369c7cd514885742c9fe08cfe85884db3e92b"
+dependencies = [
+ "aligned-vec",
+ "num-traits",
+ "wasm-bindgen",
+]
+
 [[package]]
 name = "valuable"
 version = "0.1.1"
@@ -4756,6 +5106,12 @@ version = "0.2.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
 
+[[package]]
+name = "version-compare"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "852e951cb7832cb45cb1169900d19760cfa39b82bc0ea9c0e5a14ae88411c98b"
+
 [[package]]
 name = "version_check"
 version = "0.9.5"
@@ -5324,6 +5680,15 @@ version = "0.4.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f423a2c17029964870cfaabb1f13dfab7d092a62a29a89264f4d36990ca414a"
 
+[[package]]
+name = "zune-inflate"
+version = "0.2.54"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "73ab332fe2f6680068f3582b16a24f90ad7096d5d39b974d1c0aff0125116f02"
+dependencies = [
+ "simd-adler32",
+]
+
 [[package]]
 name = "zune-jpeg"
 version = "0.4.14"

Cargo.toml

@@ -127,12 +127,13 @@ version = "0.6.2"
 default-features = false
 features = [
     "add-extension",
+    "catch-panic",
     "cors",
     "sensitive-headers",
     "set-header",
+    "timeout",
     "trace",
     "util",
-    "catch-panic",
 ]
 
 [workspace.dependencies.rustls]
@@ -178,7 +179,7 @@ version = "0.5.3"
 features = ["alloc", "rand"]
 default-features = false
 
-# Used to generate thumbnails for images
+# Used to generate thumbnails for images & blurhashes
 [workspace.dependencies.image]
 version = "0.25.5"
 default-features = false
@@ -189,6 +190,14 @@ features = [
 	"webp",
 ]
 
+[workspace.dependencies.blurhash]
+version = "0.2.3"
+default-features = false
+features = [
+	"fast-linear-to-srgb",
+	"image",
+]
+
 # logging
 [workspace.dependencies.log]
 version = "0.4.22"
@@ -333,7 +342,7 @@ version = "0.1.2"
 [workspace.dependencies.ruma]
 git = "https://github.com/girlbossceo/ruwuma"
 #branch = "conduwuit-changes"
-rev = "b560338b2a50dbf61ecfe80808b9b095ad4cec00"
+rev = "f5667c6292adb43fbe4725d31d6b5127a0cf60ce"
 features = [
     "compat",
     "rand",

arch/conduwuit.service

@@ -7,7 +7,20 @@ RequiresMountsFor=/var/lib/private/conduwuit
 
 [Service]
 DynamicUser=yes
-Type=notify
+Type=notify-reload
+ReloadSignal=SIGUSR1
+
+TTYPath=/dev/tty25
+DeviceAllow=char-tty
+StandardInput=tty-force
+StandardOutput=tty
+StandardError=journal+console
+TTYReset=yes
+# uncomment to allow buffer to be cleared every restart
+TTYVTDisallocate=no
+
+TTYColumns=120
+TTYRows=40
 
 AmbientCapabilities=
 CapabilityBoundingSet=

bin/complement

@@ -34,7 +34,8 @@ toplevel="$(git rev-parse --show-toplevel)"
 
 pushd "$toplevel" > /dev/null
 
-bin/nix-build-and-cache just .#linux-complement
+#bin/nix-build-and-cache just .#linux-complement
+bin/nix-build-and-cache just .#complement
 
 docker load < result
 popd > /dev/null

conduwuit-example.toml

@@ -377,6 +377,26 @@
 #
 #pusher_idle_timeout = 15
 
+# Maximum time to receive a request from a client (seconds).
+#
+#client_receive_timeout = 75
+
+# Maximum time to process a request received from a client (seconds).
+#
+#client_request_timeout = 180
+
+# Maximum time to transmit a response to a client (seconds)
+#
+#client_response_timeout = 120
+
+# Grace period for clean shutdown of client requests (seconds).
+#
+#client_shutdown_timeout = 10
+
+# Grace period for clean shutdown of federation requests (seconds).
+#
+#sender_shutdown_timeout = 5
+
 # Enables registration. If set to false, no users can register on this
 # server.
 #
@@ -406,8 +426,9 @@
 #
 #registration_token =
 
-# Path to a file on the system that gets read for the registration token.
-# this config option takes precedence/priority over "registration_token".
+# Path to a file on the system that gets read for additional registration
+# tokens. Multiple tokens can be added if you separate them with
+# whitespace
 #
 # conduwuit must be able to access the file, and it must not be empty
 #
@@ -1586,3 +1607,21 @@
 # This item is undocumented. Please contribute documentation for it.
 #
 #support_mxid =
+
+[global.blurhashing]
+
+# blurhashing x component, 4 is recommended by https://blurha.sh/
+#
+#components_x = 4
+
+# blurhashing y component, 3 is recommended by https://blurha.sh/
+#
+#components_y = 3
+
+# Max raw size that the server will blurhash, this is the size of the
+# image after converting it to raw data, it should be higher than the
+# upload limit but not too high. The higher it is the higher the
+# potential load will be for clients requesting blurhashes. The default
+# is 33.55MB. Setting it to 0 disables blurhashing.
+#
+#blurhash_max_raw_size = 33554432

debian/conduwuit.service

@@ -8,7 +8,20 @@ Documentation=https://conduwuit.puppyirl.gay/
 DynamicUser=yes
 User=conduwuit
 Group=conduwuit
-Type=notify
+Type=notify-reload
+ReloadSignal=SIGUSR1
+
+TTYPath=/dev/tty25
+DeviceAllow=char-tty
+StandardInput=tty-force
+StandardOutput=tty
+StandardError=journal+console
+TTYReset=yes
+# uncomment to allow buffer to be cleared every restart
+TTYVTDisallocate=no
+
+TTYColumns=120
+TTYRows=40
 
 Environment="CONDUWUIT_CONFIG=/etc/conduwuit/conduwuit.toml"
 

deps/rust-rocksdb/Cargo.toml

@@ -27,7 +27,7 @@ malloc-usable-size = ["rust-rocksdb/malloc-usable-size"]
 
 [dependencies.rust-rocksdb]
 git = "https://github.com/girlbossceo/rust-rocksdb-zaidoon1"
-rev = "1f032427d3a0e7b0f13c04b4e34712bd8610291b"
+rev = "7b0e1bbe395a41ba8a11347a4921da590e3ad0d9"
 #branch = "master"
 default-features = false
 

docs/deploying/generic.md

@@ -216,7 +216,7 @@ your server name).
 ```caddyfile
 your.server.name, your.server.name:8448 {
     # TCP reverse_proxy
-    127.0.0.1:6167
+    reverse_proxy 127.0.0.1:6167
     # UNIX socket
     #reverse_proxy unix//run/conduwuit/conduwuit.sock
 }

engage.toml

@@ -86,6 +86,7 @@ env DIRENV_DEVSHELL=all-features \
     direnv exec . \
     cargo doc \
         --workspace \
+        --locked \
         --profile test \
         --all-features \
         --no-deps \
@@ -100,8 +101,8 @@ script = """
 direnv exec . \
 cargo clippy \
     --workspace \
+    --locked \
     --profile test \
-    --all-targets \
     --color=always \
     -- \
     -D warnings
@@ -115,8 +116,8 @@ env DIRENV_DEVSHELL=all-features \
     direnv exec . \
     cargo clippy \
         --workspace \
+        --locked \
         --profile test \
-        --all-targets \
         --all-features \
         --color=always \
         -- \
@@ -124,33 +125,37 @@ env DIRENV_DEVSHELL=all-features \
 """
 
 [[task]]
-name = "clippy/jemalloc"
+name = "clippy/no-features"
+group = "lints"
+script = """
+env DIRENV_DEVSHELL=no-features \
+    direnv exec . \
+    cargo clippy \
+        --workspace \
+        --locked \
+        --profile test \
+        --no-default-features \
+        --color=always \
+        -- \
+        -D warnings
+"""
+
+[[task]]
+name = "clippy/other-features"
 group = "lints"
 script = """
 direnv exec . \
 cargo clippy \
     --workspace \
+    --locked \
     --profile test \
-    --features jemalloc \
-    --all-targets \
+    --no-default-features \
+    --features=console,systemd,element_hacks,direct_tls,perf_measurements,brotli_compression,blurhashing \
     --color=always \
     -- \
     -D warnings
 """
 
-#[[task]]
-#name = "clippy/hardened_malloc"
-#group = "lints"
-#script = """
-#cargo clippy \
-#    --workspace \
-#    --features hardened_malloc \
-#    --all-targets \
-#    --color=always \
-#    -- \
-#    -D warnings
-#"""
-
 [[task]]
 name = "lychee"
 group = "lints"
@@ -169,8 +174,10 @@ env DIRENV_DEVSHELL=all-features \
     direnv exec . \
     cargo test \
         --workspace \
+        --locked \
         --profile test \
         --all-targets \
+        --no-fail-fast \
         --all-features \
         --color=always \
         -- \
@@ -185,8 +192,28 @@ env DIRENV_DEVSHELL=default \
     direnv exec . \
     cargo test \
         --workspace \
+        --locked \
         --profile test \
         --all-targets \
+        --no-fail-fast \
+        --color=always \
+        -- \
+        --color=always
+"""
+
+[[task]]
+name = "cargo/no-features"
+group = "tests"
+script = """
+env DIRENV_DEVSHELL=no-features \
+    direnv exec . \
+    cargo test \
+        --workspace \
+        --locked \
+        --profile test \
+        --all-targets \
+        --no-fail-fast \
+        --no-default-features \
         --color=always \
         -- \
         --color=always

flake.lock

@@ -364,11 +364,11 @@
     "liburing": {
       "flake": false,
       "locked": {
-        "lastModified": 1737600516,
-        "narHash": "sha256-EKyLQ3pbcjoU5jH5atge59F4fzuhTsb6yalUj6Ve2t8=",
+        "lastModified": 1740613216,
+        "narHash": "sha256-gfjAxW5fBvPfTgkvqxs6UCzsgosJKS3Wrs0Gn8lFztk=",
         "owner": "axboe",
         "repo": "liburing",
-        "rev": "6c509e2b0c881a13b83b259a221bf15fc9b3f681",
+        "rev": "e1003e496e66f9b0ae06674869795edf772d5500",
         "type": "github"
       },
       "original": {

flake.nix

@@ -169,21 +169,9 @@
 
           # used for rust caching in CI to speed it up
           sccache
-
-          # needed so we can get rid of gcc and other unused deps that bloat OCI images
-          removeReferencesTo
         ]
         # liburing is Linux-exclusive
-        ++ lib.optional stdenv.hostPlatform.isLinux liburing
-        # needed to build Rust applications on macOS
-        ++ lib.optionals stdenv.hostPlatform.isDarwin [
-            # https://github.com/NixOS/nixpkgs/issues/206242
-            # ld: library not found for -liconv
-            libiconv
-            # https://stackoverflow.com/questions/69869574/properly-adding-darwin-apple-sdk-to-a-nix-shell
-            # https://discourse.nixos.org/t/compile-a-rust-binary-on-macos-dbcrossbar/8612
-            pkgsBuildHost.darwin.apple_sdk.frameworks.Security
-            ])
+        ++ lib.optional stdenv.hostPlatform.isLinux liburing)
         ++ scope.main.buildInputs
         ++ scope.main.propagatedBuildInputs
         ++ scope.main.nativeBuildInputs;

nix/pkgs/complement/config.toml

@@ -17,19 +17,32 @@ ip_range_denylist = []
 url_preview_domain_contains_allowlist = ["*"]
 url_preview_domain_explicit_denylist = ["*"]
 media_compat_file_link = false
-media_startup_check = false
-prune_missing_media = false
+media_startup_check = true
+prune_missing_media = true
 log_colors = false
 admin_room_notices = false
 allow_check_for_updates = false
-allow_unstable_room_versions = true
+intentionally_unknown_config_option_for_testing = true
 rocksdb_log_level = "debug"
 rocksdb_max_log_files = 1
 rocksdb_recovery_mode = 0
 rocksdb_paranoid_file_checks = true
 log_guest_registrations = false
 allow_legacy_media = true
-startup_netburst = false
+startup_netburst = true
+startup_netburst_keep = -1
+
+# valgrind makes things so slow
+dns_timeout = 60
+dns_attempts = 20
+request_conn_timeout = 60
+request_timeout = 120
+well_known_conn_timeout = 60
+well_known_timeout = 60
+federation_idle_timeout = 300
+sender_timeout = 300
+sender_idle_timeout = 300
+sender_retry_backoff_limit = 300
 
 [global.tls]
 certs = "/certificate.crt"

nix/pkgs/complement/default.nix

@@ -18,18 +18,12 @@ let
     all_features = true;
     disable_release_max_log_level = true;
     disable_features = [
-        # no reason to use jemalloc for complement, just has compatibility/build issues
-        "jemalloc"
-        "jemalloc_stats"
-        "jemalloc_prof"
         # console/CLI stuff isn't used or relevant for complement
         "console"
         "tokio_console"
         # sentry telemetry isn't useful for complement, disabled by default anyways
         "sentry_telemetry"
         "perf_measurements"
-        # the containers don't use or need systemd signal support
-        "systemd"
         # this is non-functional on nix for some reason
         "hardened_malloc"
         # dont include experimental features

nix/pkgs/main/default.nix

@@ -82,7 +82,7 @@ rust-jemalloc-sys' = (rust-jemalloc-sys.override {
 buildDepsOnlyEnv =
   let
     rocksdb' = (rocksdb.override {
-      jemalloc = rust-jemalloc-sys';
+      jemalloc = lib.optional (featureEnabled "jemalloc") rust-jemalloc-sys';
       # rocksdb fails to build with prefixed jemalloc, which is required on
       # darwin due to [1]. In this case, fall back to building rocksdb with
       # libc malloc. This should not cause conflicts, because all of the
@@ -103,6 +103,12 @@ buildDepsOnlyEnv =
       ++ [ "-DPORTABLE=haswell" ]) else ([ "-DPORTABLE=1" ])
       )
       ++ old.cmakeFlags;
+
+      # outputs has "tools" which we dont need or use
+      outputs = [ "out" ];
+
+      # preInstall hooks has stuff for messing with ldb/sst_dump which we dont need or use
+      preInstall = "";
     });
   in
   {
@@ -156,6 +162,19 @@ commonAttrs = {
       ];
     };
 
+    # This is redundant with CI
+    doCheck = false;
+
+    cargoTestCommand = "cargo test --locked ";
+    cargoExtraArgs = "--no-default-features --locked "
+      + lib.optionalString
+        (features'' != [])
+        "--features " + (builtins.concatStringsSep "," features'');
+    cargoTestExtraArgs = "--no-default-features --locked "
+      + lib.optionalString
+        (features'' != [])
+        "--features " + (builtins.concatStringsSep "," features'');
+
     dontStrip = profile == "dev" || profile == "test";
     dontPatchELF = profile == "dev" || profile == "test";
 
@@ -181,27 +200,7 @@ commonAttrs = {
       # differing values for `NIX_CFLAGS_COMPILE`, which contributes to spurious
       # rebuilds of bindgen and its depedents.
       jq
-
-      # needed so we can get rid of gcc and other unused deps that bloat OCI images
-      removeReferencesTo
-  ]
-  # needed to build Rust applications on macOS
-  ++ lib.optionals stdenv.hostPlatform.isDarwin [
-      # https://github.com/NixOS/nixpkgs/issues/206242
-      # ld: library not found for -liconv
-      libiconv
-
-      # https://stackoverflow.com/questions/69869574/properly-adding-darwin-apple-sdk-to-a-nix-shell
-      # https://discourse.nixos.org/t/compile-a-rust-binary-on-macos-dbcrossbar/8612
-      pkgsBuildHost.darwin.apple_sdk.frameworks.Security
-    ];
-
-    # for some reason gcc and other weird deps are added to OCI images and bloats it up
-    #
-    # <https://github.com/input-output-hk/haskell.nix/issues/829>
-    postInstall = with pkgsBuildHost; ''
-        find "$out" -type f -exec remove-references-to -t ${stdenv.cc} -t ${gcc} -t ${llvm} -t ${rustc.unwrapped} -t ${rustc} '{}' +
-    '';
+  ];
  };
 in
 
@@ -210,15 +209,18 @@ craneLib.buildPackage ( commonAttrs // {
     env = buildDepsOnlyEnv;
   });
 
-  cargoExtraArgs = "--no-default-features "
+  # This is redundant with CI
+  doCheck = false;
+
+  cargoTestCommand = "cargo test --locked ";
+  cargoExtraArgs = "--no-default-features --locked "
+    + lib.optionalString
+      (features'' != [])
+      "--features " + (builtins.concatStringsSep "," features'');
+  cargoTestExtraArgs = "--no-default-features --locked "
     + lib.optionalString
       (features'' != [])
       "--features " + (builtins.concatStringsSep "," features'');
-
-  # This is redundant with CI
-  cargoTestCommand = "";
-  cargoCheckCommand = "";
-  doCheck = false;
 
   env = buildPackageEnv;
 

nix/pkgs/oci-image/default.nix

@@ -36,6 +36,7 @@ dockerTools.buildLayeredImage {
       "org.opencontainers.image.documentation" = "https://conduwuit.puppyirl.gay/";
       "org.opencontainers.image.licenses" = "Apache-2.0";
       "org.opencontainers.image.revision" = inputs.self.rev or inputs.self.dirtyRev or "";
+      "org.opencontainers.image.source" = "https://github.com/girlbossceo/conduwuit";
       "org.opencontainers.image.title" = main.pname;
       "org.opencontainers.image.url" = "https://conduwuit.puppyirl.gay/";
       "org.opencontainers.image.vendor" = "girlbossceo";

src/admin/debug/commands.rs

@@ -6,8 +6,12 @@ use std::{
 };
 
 use conduwuit::{
-	debug_error, err, info, trace, utils, utils::string::EMPTY, warn, Error, PduEvent, PduId,
-	RawPduId, Result,
+	debug_error, err, info, trace, utils,
+	utils::{
+		stream::{IterStream, ReadyExt},
+		string::EMPTY,
+	},
+	warn, Error, PduEvent, PduId, RawPduId, Result,
 };
 use futures::{FutureExt, StreamExt, TryStreamExt};
 use ruma::{
@@ -54,7 +58,7 @@ pub(super) async fn get_auth_chain(
 		.rooms
 		.auth_chain
 		.event_ids_iter(room_id, once(event_id.as_ref()))
-		.await?
+		.ready_filter_map(Result::ok)
 		.count()
 		.await;
 
@@ -639,6 +643,7 @@ pub(super) async fn force_set_room_state_from_server(
 			room_id: room_id.clone().into(),
 			event_id: first_pdu.event_id.clone(),
 		})
+		.boxed()
 		.await?;
 
 	for pdu in remote_state_response.pdus.clone() {
@@ -647,6 +652,7 @@ pub(super) async fn force_set_room_state_from_server(
 			.rooms
 			.event_handler
 			.parse_incoming_pdu(&pdu)
+			.boxed()
 			.await
 		{
 			| Ok(t) => t,
@@ -710,6 +716,7 @@ pub(super) async fn force_set_room_state_from_server(
 		.rooms
 		.event_handler
 		.resolve_state(&room_id, &room_version, state)
+		.boxed()
 		.await?;
 
 	info!("Forcing new room state");
@@ -945,21 +952,57 @@ pub(super) async fn database_stats(
 	property: Option<String>,
 	map: Option<String>,
 ) -> Result<RoomMessageEventContent> {
-	let property = property.unwrap_or_else(|| "rocksdb.stats".to_owned());
 	let map_name = map.as_ref().map_or(EMPTY, String::as_str);
+	let property = property.unwrap_or_else(|| "rocksdb.stats".to_owned());
+	self.services
+		.db
+		.iter()
+		.filter(|(&name, _)| map_name.is_empty() || map_name == name)
+		.try_stream()
+		.try_for_each(|(&name, map)| {
+			let res = map.property(&property).expect("invalid property");
+			writeln!(self, "##### {name}:\n```\n{}\n```", res.trim())
+		})
+		.await?;
 
-	let mut out = String::new();
-	for (&name, map) in self.services.db.iter() {
-		if !map_name.is_empty() && map_name != name {
-			continue;
-		}
+	Ok(RoomMessageEventContent::notice_plain(""))
+}
 
-		let res = map.property(&property)?;
-		let res = res.trim();
-		writeln!(out, "##### {name}:\n```\n{res}\n```")?;
-	}
+#[admin_command]
+pub(super) async fn database_files(
+	&self,
+	map: Option<String>,
+	level: Option<i32>,
+) -> Result<RoomMessageEventContent> {
+	let mut files: Vec<_> = self.services.db.db.file_list().collect::<Result<_>>()?;
 
-	Ok(RoomMessageEventContent::notice_markdown(out))
+	files.sort_by_key(|f| f.name.clone());
+
+	writeln!(self, "| lev  | sst  | keys | dels | size | column |").await?;
+	writeln!(self, "| ---: | :--- | ---: | ---: | ---: | :---   |").await?;
+	files
+		.into_iter()
+		.filter(|file| {
+			map.as_deref()
+				.is_none_or(|map| map == file.column_family_name)
+		})
+		.filter(|file| level.as_ref().is_none_or(|&level| level == file.level))
+		.try_stream()
+		.try_for_each(|file| {
+			writeln!(
+				self,
+				"| {} | {:<13} | {:7}+ | {:4}- | {:9} | {} |",
+				file.level,
+				file.name,
+				file.num_entries,
+				file.num_deletions,
+				file.size,
+				file.column_family_name,
+			)
+		})
+		.await?;
+
+	Ok(RoomMessageEventContent::notice_plain(""))
 }
 
 #[admin_command]

src/admin/debug/mod.rs

@@ -226,6 +226,14 @@ pub(super) enum DebugCommand {
 	/// - Trim memory usage
 	TrimMemory,
 
+	/// - List database files
+	DatabaseFiles {
+		map: Option<String>,
+
+		#[arg(long)]
+		level: Option<i32>,
+	},
+
 	/// - Developer test stubs
 	#[command(subcommand)]
 	#[allow(non_snake_case)]

src/admin/query/account_data.rs

@@ -41,7 +41,7 @@ async fn changes_since(
 	let results: Vec<_> = self
 		.services
 		.account_data
-		.changes_since(room_id.as_deref(), &user_id, since)
+		.changes_since(room_id.as_deref(), &user_id, since, None)
 		.collect()
 		.await;
 	let query_time = timer.elapsed();

src/admin/query/users.rs

@@ -413,7 +413,7 @@ async fn get_to_device_events(
 	let result = self
 		.services
 		.users
-		.get_to_device_events(&user_id, &device_id)
+		.get_to_device_events(&user_id, &device_id, None, None)
 		.collect::<Vec<_>>()
 		.await;
 	let query_time = timer.elapsed();

src/admin/room/alias.rs

@@ -72,7 +72,7 @@ pub(super) async fn reprocess(
 					))),
 			};
 			match command {
-				| RoomAliasCommand::Set { force, room_id, .. } =>
+				| RoomAliasCommand::Set { force, room_id, .. } => {
 					match (force, services.rooms.alias.resolve_local_alias(&room_alias).await) {
 						| (true, Ok(id)) => {
 							match services.rooms.alias.set_alias(
@@ -106,8 +106,9 @@ pub(super) async fn reprocess(
 								))),
 							}
 						},
-					},
-				| RoomAliasCommand::Remove { .. } =>
+					}
+				},
+				| RoomAliasCommand::Remove { .. } => {
 					match services.rooms.alias.resolve_local_alias(&room_alias).await {
 						| Ok(id) => match services
 							.rooms
@@ -124,15 +125,17 @@ pub(super) async fn reprocess(
 						},
 						| Err(_) =>
 							Ok(RoomMessageEventContent::text_plain("Alias isn't in use.")),
-					},
-				| RoomAliasCommand::Which { .. } =>
+					}
+				},
+				| RoomAliasCommand::Which { .. } => {
 					match services.rooms.alias.resolve_local_alias(&room_alias).await {
 						| Ok(id) => Ok(RoomMessageEventContent::text_plain(format!(
 							"Alias resolves to {id}"
 						))),
 						| Err(_) =>
 							Ok(RoomMessageEventContent::text_plain("Alias isn't in use.")),
-					},
+					}
+				},
 				| RoomAliasCommand::List { .. } => unreachable!(),
 			}
 		},

src/admin/server/commands.rs

@@ -92,7 +92,7 @@ pub(super) async fn clear_caches(&self) -> Result<RoomMessageEventContent> {
 
 #[admin_command]
 pub(super) async fn list_backups(&self) -> Result<RoomMessageEventContent> {
-	let result = self.services.globals.db.backup_list()?;
+	let result = self.services.db.db.backup_list()?;
 
 	if result.is_empty() {
 		Ok(RoomMessageEventContent::text_plain("No backups found."))
@@ -103,31 +103,24 @@ pub(super) async fn list_backups(&self) -> Result<RoomMessageEventContent> {
 
 #[admin_command]
 pub(super) async fn backup_database(&self) -> Result<RoomMessageEventContent> {
-	let globals = Arc::clone(&self.services.globals);
+	let db = Arc::clone(&self.services.db);
 	let mut result = self
 		.services
 		.server
 		.runtime()
-		.spawn_blocking(move || match globals.db.backup() {
+		.spawn_blocking(move || match db.db.backup() {
 			| Ok(()) => String::new(),
 			| Err(e) => e.to_string(),
 		})
 		.await?;
 
 	if result.is_empty() {
-		result = self.services.globals.db.backup_list()?;
+		result = self.services.db.db.backup_list()?;
 	}
 
 	Ok(RoomMessageEventContent::notice_markdown(result))
 }
 
-#[admin_command]
-pub(super) async fn list_database_files(&self) -> Result<RoomMessageEventContent> {
-	let result = self.services.globals.db.file_list()?;
-
-	Ok(RoomMessageEventContent::notice_markdown(result))
-}
-
 #[admin_command]
 pub(super) async fn admin_notice(&self, message: Vec<String>) -> Result<RoomMessageEventContent> {
 	let message = message.join(" ");

src/admin/server/mod.rs

@@ -46,9 +46,6 @@ pub(super) enum ServerCommand {
 	/// - List database backups
 	ListBackups,
 
-	/// - List database files
-	ListDatabaseFiles,
-
 	/// - Send a message to the admin room.
 	AdminNotice {
 		message: Vec<String>,

src/api/client/media.rs

@@ -57,19 +57,28 @@ pub(crate) async fn create_content_route(
 	let filename = body.filename.as_deref();
 	let content_type = body.content_type.as_deref();
 	let content_disposition = make_content_disposition(None, content_type, filename);
-	let mxc = Mxc {
+	let ref mxc = Mxc {
 		server_name: services.globals.server_name(),
 		media_id: &utils::random_string(MXC_LENGTH),
 	};
 
 	services
 		.media
-		.create(&mxc, Some(user), Some(&content_disposition), content_type, &body.file)
-		.await
-		.map(|()| create_content::v3::Response {
-			content_uri: mxc.to_string().into(),
-			blurhash: None,
-		})
+		.create(mxc, Some(user), Some(&content_disposition), content_type, &body.file)
+		.await?;
+
+	let blurhash = body.generate_blurhash.then(|| {
+		services
+			.media
+			.create_blurhash(&body.file, content_type, filename)
+			.ok()
+			.flatten()
+	});
+
+	Ok(create_content::v3::Response {
+		content_uri: mxc.to_string().into(),
+		blurhash: blurhash.flatten(),
+	})
 }
 
 /// # `GET /_matrix/client/v1/media/thumbnail/{serverName}/{mediaId}`

src/api/client/membership.rs

@@ -1,6 +1,7 @@
 use std::{
 	borrow::Borrow,
 	collections::{BTreeMap, HashMap, HashSet},
+	iter::once,
 	net::IpAddr,
 	sync::Arc,
 };
@@ -45,7 +46,10 @@ use ruma::{
 use service::{
 	appservice::RegistrationInfo,
 	pdu::gen_event_id,
-	rooms::{state::RoomMutexGuard, state_compressor::HashSetCompressStateEvent},
+	rooms::{
+		state::RoomMutexGuard,
+		state_compressor::{CompressedState, HashSetCompressStateEvent},
+	},
 	Services,
 };
 
@@ -1168,7 +1172,7 @@ async fn join_room_by_id_helper_remote(
 	}
 
 	info!("Compressing state from send_join");
-	let compressed: HashSet<_> = services
+	let compressed: CompressedState = services
 		.rooms
 		.state_compressor
 		.compress_state_events(state.iter().map(|(ssk, eid)| (ssk, eid.borrow())))
@@ -1216,7 +1220,7 @@ async fn join_room_by_id_helper_remote(
 		.append_pdu(
 			&parsed_join_pdu,
 			join_event,
-			vec![(*parsed_join_pdu.event_id).to_owned()],
+			once(parsed_join_pdu.event_id.borrow()),
 			&state_lock,
 		)
 		.await?;
@@ -2195,7 +2199,7 @@ async fn knock_room_helper_local(
 		.append_pdu(
 			&parsed_knock_pdu,
 			knock_event,
-			vec![(*parsed_knock_pdu.event_id).to_owned()],
+			once(parsed_knock_pdu.event_id.borrow()),
 			&state_lock,
 		)
 		.await?;
@@ -2339,7 +2343,7 @@ async fn knock_room_helper_remote(
 	}
 
 	info!("Compressing state from send_knock");
-	let compressed: HashSet<_> = services
+	let compressed: CompressedState = services
 		.rooms
 		.state_compressor
 		.compress_state_events(state_map.iter().map(|(ssk, eid)| (ssk, eid.borrow())))
@@ -2394,7 +2398,7 @@ async fn knock_room_helper_remote(
 		.append_pdu(
 			&parsed_knock_pdu,
 			knock_event,
-			vec![(*parsed_knock_pdu.event_id).to_owned()],
+			once(parsed_knock_pdu.event_id.borrow()),
 			&state_lock,
 		)
 		.await?;

src/api/client/sync/v3.rs

@@ -1,22 +1,22 @@
 use std::{
 	cmp::{self},
-	collections::{hash_map::Entry, BTreeMap, HashMap, HashSet},
+	collections::{BTreeMap, HashMap, HashSet},
 	time::Duration,
 };
 
 use axum::extract::State;
 use conduwuit::{
-	at, err, error, extract_variant, is_equal_to,
-	pdu::EventHash,
+	at, err, error, extract_variant, is_equal_to, pair_of,
+	pdu::{Event, EventHash},
+	ref_at,
 	result::FlatOk,
 	utils::{
 		self,
-		future::OptionExt,
 		math::ruma_from_u64,
-		stream::{BroadbandExt, Tools, WidebandExt},
+		stream::{BroadbandExt, Tools, TryExpect, WidebandExt},
 		BoolExt, IterStream, ReadyExt, TryFutureExtExt,
 	},
-	Error, PduCount, PduEvent, Result,
+	PduCount, PduEvent, Result,
 };
 use conduwuit_service::{
 	rooms::{
@@ -45,7 +45,7 @@ use ruma::{
 		uiaa::UiaaResponse,
 	},
 	events::{
-		presence::PresenceEvent,
+		presence::{PresenceEvent, PresenceEventContent},
 		room::member::{MembershipState, RoomMemberEventContent},
 		AnyRawAccountDataEvent, AnySyncEphemeralRoomEvent, StateEventType,
 		TimelineEventType::*,
@@ -53,6 +53,7 @@ use ruma::{
 	serde::Raw,
 	uint, DeviceId, EventId, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, UserId,
 };
+use service::rooms::short::{ShortEventId, ShortStateKey};
 
 use super::{load_timeline, share_encrypted_room};
 use crate::{client::ignored_filter, Ruma, RumaResponse};
@@ -62,11 +63,12 @@ struct StateChanges {
 	heroes: Option<Vec<OwnedUserId>>,
 	joined_member_count: Option<u64>,
 	invited_member_count: Option<u64>,
-	joined_since_last_sync: bool,
 	state_events: Vec<PduEvent>,
+	device_list_updates: HashSet<OwnedUserId>,
+	left_encrypted_users: HashSet<OwnedUserId>,
 }
 
-type PresenceUpdates = HashMap<OwnedUserId, PresenceEvent>;
+type PresenceUpdates = HashMap<OwnedUserId, PresenceEventContent>;
 
 /// # `GET /_matrix/client/r0/sync`
 ///
@@ -285,20 +287,20 @@ pub(crate) async fn build_sync_events(
 
 	let account_data = services
 		.account_data
-		.changes_since(None, sender_user, since)
+		.changes_since(None, sender_user, since, Some(next_batch))
 		.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
 		.collect();
 
 	// Look for device list updates of this account
 	let keys_changed = services
 		.users
-		.keys_changed(sender_user, since, None)
+		.keys_changed(sender_user, since, Some(next_batch))
 		.map(ToOwned::to_owned)
 		.collect::<HashSet<_>>();
 
 	let to_device_events = services
 		.users
-		.get_to_device_events(sender_user, sender_device)
+		.get_to_device_events(sender_user, sender_device, Some(since), Some(next_batch))
 		.collect::<Vec<_>>();
 
 	let device_one_time_keys_count = services
@@ -325,18 +327,16 @@ pub(crate) async fn build_sync_events(
 
 	// If the user doesn't share an encrypted room with the target anymore, we need
 	// to tell them
-	let device_list_left = left_encrypted_users
+	let device_list_left: HashSet<_> = left_encrypted_users
 		.into_iter()
 		.stream()
 		.broad_filter_map(|user_id| async move {
-			let no_shared_encrypted_room =
-				!share_encrypted_room(services, sender_user, &user_id, None).await;
-			no_shared_encrypted_room.then_some(user_id)
-		})
-		.ready_fold(HashSet::new(), |mut device_list_left, user_id| {
-			device_list_left.insert(user_id);
-			device_list_left
+			share_encrypted_room(services, sender_user, &user_id, None)
+				.await
+				.eq(&false)
+				.then_some(user_id)
 		})
+		.collect()
 		.await;
 
 	let response = sync_events::v3::Response {
@@ -351,9 +351,11 @@ pub(crate) async fn build_sync_events(
 		next_batch: next_batch.to_string(),
 		presence: Presence {
 			events: presence_updates
-				.unwrap_or_default()
-				.into_values()
-				.map(|v| Raw::new(&v).expect("PresenceEvent always serializes successfully"))
+				.into_iter()
+				.flat_map(IntoIterator::into_iter)
+				.map(|(sender, content)| PresenceEvent { content, sender })
+				.map(|ref event| Raw::new(event))
+				.filter_map(Result::ok)
 				.collect(),
 		},
 		rooms: Rooms {
@@ -390,45 +392,8 @@ async fn process_presence_updates(
 				.map_ok(move |event| (user_id, event))
 				.ok()
 		})
-		.ready_fold(PresenceUpdates::new(), |mut updates, (user_id, event)| {
-			match updates.entry(user_id.into()) {
-				| Entry::Vacant(slot) => {
-					let mut new_event = event;
-					new_event.content.last_active_ago = match new_event.content.currently_active {
-						| Some(true) => None,
-						| _ => new_event.content.last_active_ago,
-					};
-
-					slot.insert(new_event);
-				},
-				| Entry::Occupied(mut slot) => {
-					let curr_event = slot.get_mut();
-					let curr_content = &mut curr_event.content;
-					let new_content = event.content;
-
-					// Update existing presence event with more info
-					curr_content.presence = new_content.presence;
-					curr_content.status_msg = new_content
-						.status_msg
-						.or_else(|| curr_content.status_msg.take());
-					curr_content.displayname = new_content
-						.displayname
-						.or_else(|| curr_content.displayname.take());
-					curr_content.avatar_url = new_content
-						.avatar_url
-						.or_else(|| curr_content.avatar_url.take());
-					curr_content.currently_active = new_content
-						.currently_active
-						.or(curr_content.currently_active);
-					curr_content.last_active_ago = match curr_content.currently_active {
-						| Some(true) => None,
-						| _ => new_content.last_active_ago.or(curr_content.last_active_ago),
-					};
-				},
-			};
-
-			updates
-		})
+		.map(|(user_id, event)| (user_id.to_owned(), event.content))
+		.collect()
 		.await
 }
 
@@ -657,6 +622,40 @@ async fn load_joined_room(
 			.await?;
 
 	let (timeline_pdus, limited) = timeline;
+	let initial = since_shortstatehash.is_none();
+	let lazy_loading_enabled = filter.room.state.lazy_load_options.is_enabled()
+		|| filter.room.timeline.lazy_load_options.is_enabled();
+
+	let lazy_loading_context = &lazy_loading::Context {
+		user_id: sender_user,
+		device_id: sender_device,
+		room_id,
+		token: Some(since),
+		options: Some(&filter.room.state.lazy_load_options),
+	};
+
+	// Reset lazy loading because this is an initial sync
+	let lazy_load_reset: OptionFuture<_> = initial
+		.then(|| services.rooms.lazy_loading.reset(lazy_loading_context))
+		.into();
+
+	lazy_load_reset.await;
+	let witness: OptionFuture<_> = lazy_loading_enabled
+		.then(|| {
+			let witness: Witness = timeline_pdus
+				.iter()
+				.map(ref_at!(1))
+				.map(Event::sender)
+				.map(Into::into)
+				.chain(receipt_events.keys().map(Into::into))
+				.collect();
+
+			services
+				.rooms
+				.lazy_loading
+				.witness_retain(witness, lazy_loading_context)
+		})
+		.into();
 
 	let last_notification_read: OptionFuture<_> = timeline_pdus
 		.is_empty()
@@ -668,10 +667,6 @@ async fn load_joined_room(
 		})
 		.into();
 
-	let no_state_changes = timeline_pdus.is_empty()
-		&& (since_shortstatehash.is_none()
-			|| since_shortstatehash.is_some_and(is_equal_to!(current_shortstatehash)));
-
 	let since_sender_member: OptionFuture<_> = since_shortstatehash
 		.map(|short| {
 			services
@@ -682,125 +677,85 @@ async fn load_joined_room(
 		})
 		.into();
 
+	let (last_notification_read, since_sender_member, witness) =
+		join3(last_notification_read, since_sender_member, witness).await;
+
 	let joined_since_last_sync =
 		since_sender_member
-			.await
 			.flatten()
 			.is_none_or(|content: RoomMemberEventContent| {
 				content.membership != MembershipState::Join
 			});
 
-	let lazy_loading_enabled = filter.room.state.lazy_load_options.is_enabled()
-		|| filter.room.timeline.lazy_load_options.is_enabled();
-
-	let generate_witness =
-		lazy_loading_enabled && (since_shortstatehash.is_none() || joined_since_last_sync);
-
-	let lazy_reset = lazy_loading_enabled && since_shortstatehash.is_none();
-
-	let lazy_loading_context = &lazy_loading::Context {
-		user_id: sender_user,
-		device_id: sender_device,
-		room_id,
-		token: None,
-		options: Some(&filter.room.state.lazy_load_options),
-	};
-
-	// Reset lazy loading because this is an initial sync
-	let lazy_load_reset: OptionFuture<_> = lazy_reset
-		.then(|| services.rooms.lazy_loading.reset(lazy_loading_context))
-		.into();
-
-	lazy_load_reset.await;
-	let witness: Option<Witness> = generate_witness.then(|| {
-		timeline_pdus
-			.iter()
-			.map(|(_, pdu)| pdu.sender.clone())
-			.chain(receipt_events.keys().cloned())
-			.collect()
-	});
-
-	let witness: OptionFuture<_> = witness
-		.map(|witness| {
-			services
-				.rooms
-				.lazy_loading
-				.witness_retain(witness, lazy_loading_context)
-		})
-		.into();
-
-	let witness = witness.await;
-	let mut device_list_updates = HashSet::<OwnedUserId>::new();
-	let mut left_encrypted_users = HashSet::<OwnedUserId>::new();
 	let StateChanges {
 		heroes,
 		joined_member_count,
 		invited_member_count,
+		mut state_events,
+		mut device_list_updates,
+		left_encrypted_users,
+	} = calculate_state_changes(
+		services,
+		sender_user,
+		room_id,
+		full_state,
+		filter,
+		since_shortstatehash,
+		current_shortstatehash,
 		joined_since_last_sync,
-		state_events,
-	} = if no_state_changes {
-		StateChanges::default()
-	} else {
-		calculate_state_changes(
-			services,
-			sender_user,
-			room_id,
-			full_state,
-			filter,
-			&mut device_list_updates,
-			&mut left_encrypted_users,
-			since_shortstatehash,
-			current_shortstatehash,
-			joined_since_last_sync,
-			witness.as_ref(),
-		)
-		.boxed()
-		.await?
+		witness.as_ref(),
+	)
+	.boxed()
+	.await?;
+
+	let is_sender_membership = |pdu: &PduEvent| {
+		pdu.kind == StateEventType::RoomMember.into()
+			&& pdu
+				.state_key
+				.as_deref()
+				.is_some_and(is_equal_to!(sender_user.as_str()))
 	};
 
+	let joined_sender_member: Option<_> = (joined_since_last_sync && timeline_pdus.is_empty())
+		.then(|| {
+			state_events
+				.iter()
+				.position(is_sender_membership)
+				.map(|pos| state_events.swap_remove(pos))
+		})
+		.flatten();
+
+	let prev_batch = timeline_pdus.first().map(at!(0)).or_else(|| {
+		joined_sender_member
+			.is_some()
+			.then_some(since)
+			.map(Into::into)
+	});
+
+	let room_events = timeline_pdus
+		.into_iter()
+		.stream()
+		.wide_filter_map(|item| ignored_filter(services, item, sender_user))
+		.map(at!(1))
+		.chain(joined_sender_member.into_iter().stream())
+		.map(|pdu| pdu.to_sync_room_event())
+		.collect::<Vec<_>>();
+
 	let account_data_events = services
 		.account_data
-		.changes_since(Some(room_id), sender_user, since)
+		.changes_since(Some(room_id), sender_user, since, Some(next_batch))
 		.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
 		.collect();
 
 	// Look for device list updates in this room
 	let device_updates = services
 		.users
-		.room_keys_changed(room_id, since, None)
+		.room_keys_changed(room_id, since, Some(next_batch))
 		.map(|(user_id, _)| user_id)
 		.map(ToOwned::to_owned)
 		.collect::<Vec<_>>();
 
-	let room_events = timeline_pdus
-		.iter()
-		.stream()
-		.wide_filter_map(|item| ignored_filter(services, item.clone(), sender_user))
-		.map(|(_, pdu)| pdu.to_sync_room_event())
-		.collect();
-
-	let typing_events = services
-		.rooms
-		.typing
-		.last_typing_update(room_id)
-		.and_then(|count| async move {
-			if count <= since {
-				return Ok(Vec::<Raw<AnySyncEphemeralRoomEvent>>::new());
-			}
-
-			let typings = services
-				.rooms
-				.typing
-				.typings_all(room_id, sender_user)
-				.await?;
-
-			Ok(vec![serde_json::from_str(&serde_json::to_string(&typings)?)?])
-		})
-		.unwrap_or(Vec::new());
-
-	let send_notification_counts = last_notification_read
-		.is_none_or(|&count| count > since)
-		.await;
+	let send_notification_counts = last_notification_read.is_none_or(|count| count > since);
 
 	let notification_count: OptionFuture<_> = send_notification_counts
 		.then(|| {
@@ -824,8 +779,27 @@ async fn load_joined_room(
 		})
 		.into();
 
-	let events = join3(room_events, account_data_events, typing_events);
+	let typing_events = services
+		.rooms
+		.typing
+		.last_typing_update(room_id)
+		.and_then(|count| async move {
+			if count <= since {
+				return Ok(Vec::<Raw<AnySyncEphemeralRoomEvent>>::new());
+			}
+
+			let typings = services
+				.rooms
+				.typing
+				.typings_all(room_id, sender_user)
+				.await?;
+
+			Ok(vec![serde_json::from_str(&serde_json::to_string(&typings)?)?])
+		})
+		.unwrap_or(Vec::new());
+
 	let unread_notifications = join(notification_count, highlight_count);
+	let events = join3(room_events, account_data_events, typing_events);
 	let (unread_notifications, events, device_updates) =
 		join3(unread_notifications, events, device_updates)
 			.boxed()
@@ -882,12 +856,8 @@ async fn load_joined_room(
 		unread_notifications: UnreadNotificationsCount { highlight_count, notification_count },
 		timeline: Timeline {
 			limited: limited || joined_since_last_sync,
+			prev_batch: prev_batch.as_ref().map(ToString::to_string),
 			events: room_events,
-			prev_batch: timeline_pdus
-				.first()
-				.map(at!(0))
-				.as_ref()
-				.map(ToString::to_string),
 		},
 		state: RoomState {
 			events: state_events
@@ -919,14 +889,12 @@ async fn calculate_state_changes(
 	room_id: &RoomId,
 	full_state: bool,
 	filter: &FilterDefinition,
-	device_list_updates: &mut HashSet<OwnedUserId>,
-	left_encrypted_users: &mut HashSet<OwnedUserId>,
 	since_shortstatehash: Option<ShortStateHash>,
 	current_shortstatehash: ShortStateHash,
 	joined_since_last_sync: bool,
 	witness: Option<&Witness>,
 ) -> Result<StateChanges> {
-	if since_shortstatehash.is_none() || joined_since_last_sync {
+	if since_shortstatehash.is_none() {
 		calculate_state_initial(
 			services,
 			sender_user,
@@ -944,11 +912,10 @@ async fn calculate_state_changes(
 			room_id,
 			full_state,
 			filter,
-			device_list_updates,
-			left_encrypted_users,
 			since_shortstatehash,
 			current_shortstatehash,
 			joined_since_last_sync,
+			witness,
 		)
 		.await
 	}
@@ -961,7 +928,7 @@ async fn calculate_state_initial(
 	sender_user: &UserId,
 	room_id: &RoomId,
 	full_state: bool,
-	filter: &FilterDefinition,
+	_filter: &FilterDefinition,
 	current_shortstatehash: ShortStateHash,
 	witness: Option<&Witness>,
 ) -> Result<StateChanges> {
@@ -979,20 +946,14 @@ async fn calculate_state_initial(
 		.zip(event_ids.into_iter().stream())
 		.ready_filter_map(|item| Some((item.0.ok()?, item.1)))
 		.ready_filter_map(|((event_type, state_key), event_id)| {
-			let lazy_load_enabled = filter.room.state.lazy_load_options.is_enabled()
-				|| filter.room.timeline.lazy_load_options.is_enabled();
-
-			if lazy_load_enabled
+			let lazy = !full_state
 				&& event_type == StateEventType::RoomMember
-				&& !full_state
 				&& state_key.as_str().try_into().is_ok_and(|user_id: &UserId| {
 					sender_user != user_id
 						&& witness.is_some_and(|witness| !witness.contains(user_id))
-				}) {
-				return None;
-			}
+				});
 
-			Some(event_id)
+			lazy.or_some(event_id)
 		})
 		.broad_filter_map(|event_id: OwnedEventId| async move {
 			services.rooms.timeline.get_pdu(&event_id).await.ok()
@@ -1011,130 +972,170 @@ async fn calculate_state_initial(
 		heroes,
 		joined_member_count,
 		invited_member_count,
-		joined_since_last_sync: true,
 		state_events,
+		..Default::default()
 	})
 }
 
 #[tracing::instrument(name = "incremental", level = "trace", skip_all)]
 #[allow(clippy::too_many_arguments)]
-async fn calculate_state_incremental(
+async fn calculate_state_incremental<'a>(
 	services: &Services,
-	sender_user: &UserId,
+	sender_user: &'a UserId,
 	room_id: &RoomId,
 	full_state: bool,
 	_filter: &FilterDefinition,
-	device_list_updates: &mut HashSet<OwnedUserId>,
-	left_encrypted_users: &mut HashSet<OwnedUserId>,
 	since_shortstatehash: Option<ShortStateHash>,
 	current_shortstatehash: ShortStateHash,
 	joined_since_last_sync: bool,
+	witness: Option<&'a Witness>,
 ) -> Result<StateChanges> {
-	// Incremental /sync
-	let since_shortstatehash =
-		since_shortstatehash.expect("missing since_shortstatehash on incremental sync");
+	let since_shortstatehash = since_shortstatehash.unwrap_or(current_shortstatehash);
 
-	let mut delta_state_events = Vec::new();
-
-	if since_shortstatehash != current_shortstatehash {
-		let current_state_ids = services
-			.rooms
-			.state_accessor
-			.state_full_ids(current_shortstatehash)
-			.collect();
-
-		let since_state_ids = services
-			.rooms
-			.state_accessor
-			.state_full_ids(since_shortstatehash)
-			.collect();
-
-		let (current_state_ids, since_state_ids): (
-			HashMap<_, OwnedEventId>,
-			HashMap<_, OwnedEventId>,
-		) = join(current_state_ids, since_state_ids).await;
-
-		current_state_ids
-			.iter()
-			.stream()
-			.ready_filter(|(key, id)| full_state || since_state_ids.get(key) != Some(id))
-			.wide_filter_map(|(_, id)| services.rooms.timeline.get_pdu(id).ok())
-			.ready_for_each(|pdu| delta_state_events.push(pdu))
-			.await;
-	}
+	let state_changed = since_shortstatehash != current_shortstatehash;
 
 	let encrypted_room = services
 		.rooms
 		.state_accessor
 		.state_get(current_shortstatehash, &StateEventType::RoomEncryption, "")
-		.is_ok();
+		.is_ok()
+		.await;
 
-	let since_encryption = services
-		.rooms
-		.state_accessor
-		.state_get(since_shortstatehash, &StateEventType::RoomEncryption, "")
-		.is_ok();
-
-	let (encrypted_room, since_encryption) = join(encrypted_room, since_encryption).await;
-
-	// Calculations:
-	let new_encrypted_room = encrypted_room && !since_encryption;
-
-	let send_member_count = delta_state_events
-		.iter()
-		.any(|event| event.kind == RoomMember);
-
-	if encrypted_room {
-		for state_event in &delta_state_events {
-			if state_event.kind != RoomMember {
-				continue;
-			}
-
-			if let Some(state_key) = &state_event.state_key {
-				let user_id = UserId::parse(state_key)
-					.map_err(|_| Error::bad_database("Invalid UserId in member PDU."))?;
-
-				if user_id == sender_user {
-					continue;
-				}
-
-				let content: RoomMemberEventContent = state_event.get_content()?;
-
-				match content.membership {
-					| MembershipState::Join => {
-						// A new user joined an encrypted room
-						if !share_encrypted_room(services, sender_user, user_id, Some(room_id))
-							.await
-						{
-							device_list_updates.insert(user_id.into());
-						}
-					},
-					| MembershipState::Leave => {
-						// Write down users that have left encrypted rooms we are in
-						left_encrypted_users.insert(user_id.into());
-					},
-					| _ => {},
-				}
-			}
-		}
-	}
-
-	if joined_since_last_sync && encrypted_room || new_encrypted_room {
-		let updates: Vec<OwnedUserId> = services
+	let state_get_shorteventid = |user_id: &'a UserId| {
+		services
 			.rooms
-			.state_cache
-			.room_members(room_id)
-			.ready_filter(|user_id| sender_user != *user_id)
-			.filter_map(|user_id| {
-				share_encrypted_room(services, sender_user, user_id, Some(room_id))
-					.map(|res| res.or_some(user_id.to_owned()))
-			})
-			.collect()
-			.await;
+			.state_accessor
+			.state_get_shortid(
+				current_shortstatehash,
+				&StateEventType::RoomMember,
+				user_id.as_str(),
+			)
+			.ok()
+	};
 
-		// If the user is in a new encrypted room, give them all joined users
-		device_list_updates.extend(updates);
-	}
+	let lazy_state_ids: OptionFuture<_> = witness
+		.filter(|_| !full_state && !encrypted_room)
+		.map(|witness| {
+			witness
+				.iter()
+				.stream()
+				.broad_filter_map(|user_id| state_get_shorteventid(user_id))
+				.into_future()
+		})
+		.into();
+
+	let state_diff: OptionFuture<_> = (!full_state && state_changed)
+		.then(|| {
+			services
+				.rooms
+				.state_accessor
+				.state_added((since_shortstatehash, current_shortstatehash))
+				.boxed()
+				.into_future()
+		})
+		.into();
+
+	let current_state_ids: OptionFuture<_> = full_state
+		.then(|| {
+			services
+				.rooms
+				.state_accessor
+				.state_full_shortids(current_shortstatehash)
+				.expect_ok()
+				.boxed()
+				.into_future()
+		})
+		.into();
+
+	let lazy_state_ids = lazy_state_ids
+		.map(|opt| {
+			opt.map(|(curr, next)| {
+				let opt = curr;
+				let iter = Option::into_iter(opt);
+				IterStream::stream(iter).chain(next)
+			})
+		})
+		.map(Option::into_iter)
+		.map(IterStream::stream)
+		.flatten_stream()
+		.flatten();
+
+	let state_diff_ids = state_diff
+		.map(|opt| {
+			opt.map(|(curr, next)| {
+				let opt = curr;
+				let iter = Option::into_iter(opt);
+				IterStream::stream(iter).chain(next)
+			})
+		})
+		.map(Option::into_iter)
+		.map(IterStream::stream)
+		.flatten_stream()
+		.flatten();
+
+	let state_events = current_state_ids
+		.map(|opt| {
+			opt.map(|(curr, next)| {
+				let opt = curr;
+				let iter = Option::into_iter(opt);
+				IterStream::stream(iter).chain(next)
+			})
+		})
+		.map(Option::into_iter)
+		.map(IterStream::stream)
+		.flatten_stream()
+		.flatten()
+		.chain(state_diff_ids)
+		.broad_filter_map(|(shortstatekey, shorteventid)| async move {
+			if witness.is_none() || encrypted_room {
+				return Some(shorteventid);
+			}
+
+			lazy_filter(services, sender_user, shortstatekey, shorteventid).await
+		})
+		.chain(lazy_state_ids)
+		.broad_filter_map(|shorteventid| {
+			services
+				.rooms
+				.short
+				.get_eventid_from_short(shorteventid)
+				.ok()
+		})
+		.broad_filter_map(|event_id: OwnedEventId| async move {
+			services.rooms.timeline.get_pdu(&event_id).await.ok()
+		})
+		.collect::<Vec<_>>()
+		.await;
+
+	let (device_list_updates, left_encrypted_users) = state_events
+		.iter()
+		.stream()
+		.ready_filter(|_| encrypted_room)
+		.ready_filter(|state_event| state_event.kind == RoomMember)
+		.ready_filter_map(|state_event| {
+			let content: RoomMemberEventContent = state_event.get_content().ok()?;
+			let user_id: OwnedUserId = state_event.state_key.as_ref()?.parse().ok()?;
+
+			Some((content, user_id))
+		})
+		.fold_default(|(mut dlu, mut leu): pair_of!(HashSet<_>), (content, user_id)| async move {
+			use MembershipState::*;
+
+			let shares_encrypted_room =
+				|user_id| share_encrypted_room(services, sender_user, user_id, Some(room_id));
+
+			match content.membership {
+				| Leave => leu.insert(user_id),
+				| Join if joined_since_last_sync || !shares_encrypted_room(&user_id).await =>
+					dlu.insert(user_id),
+				| _ => false,
+			};
+
+			(dlu, leu)
+		})
+		.await;
+
+	let send_member_count = state_events.iter().any(|event| event.kind == RoomMember);
 
 	let (joined_member_count, invited_member_count, heroes) = if send_member_count {
 		calculate_counts(services, room_id, sender_user).await?
@@ -1146,11 +1147,29 @@ async fn calculate_state_incremental(
 		heroes,
 		joined_member_count,
 		invited_member_count,
-		joined_since_last_sync,
-		state_events: delta_state_events,
+		state_events,
+		device_list_updates,
+		left_encrypted_users,
 	})
 }
 
+async fn lazy_filter(
+	services: &Services,
+	sender_user: &UserId,
+	shortstatekey: ShortStateKey,
+	shorteventid: ShortEventId,
+) -> Option<ShortEventId> {
+	let (event_type, state_key) = services
+		.rooms
+		.short
+		.get_statekey_from_short(shortstatekey)
+		.await
+		.ok()?;
+
+	(event_type != StateEventType::RoomMember || state_key == sender_user.as_str())
+		.then_some(shorteventid)
+}
+
 async fn calculate_counts(
 	services: &Services,
 	room_id: &RoomId,

src/api/client/sync/v4.rs

@@ -153,7 +153,7 @@ pub(crate) async fn sync_events_v4_route(
 	if body.extensions.account_data.enabled.unwrap_or(false) {
 		account_data.global = services
 			.account_data
-			.changes_since(None, sender_user, globalsince)
+			.changes_since(None, sender_user, globalsince, Some(next_batch))
 			.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
 			.collect()
 			.await;
@@ -164,7 +164,7 @@ pub(crate) async fn sync_events_v4_route(
 					room.clone(),
 					services
 						.account_data
-						.changes_since(Some(&room), sender_user, globalsince)
+						.changes_since(Some(&room), sender_user, globalsince, Some(next_batch))
 						.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
 						.collect()
 						.await,
@@ -531,7 +531,7 @@ pub(crate) async fn sync_events_v4_route(
 			room_id.to_owned(),
 			services
 				.account_data
-				.changes_since(Some(room_id), sender_user, *roomsince)
+				.changes_since(Some(room_id), sender_user, *roomsince, Some(next_batch))
 				.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
 				.collect()
 				.await,
@@ -779,7 +779,12 @@ pub(crate) async fn sync_events_v4_route(
 				Some(sync_events::v4::ToDevice {
 					events: services
 						.users
-						.get_to_device_events(sender_user, &sender_device)
+						.get_to_device_events(
+							sender_user,
+							&sender_device,
+							Some(globalsince),
+							Some(next_batch),
+						)
 						.collect()
 						.await,
 					next_batch: next_batch.to_string(),

src/api/client/sync/v5.rs

@@ -390,7 +390,7 @@ async fn process_rooms(
 				room_id.to_owned(),
 				services
 					.account_data
-					.changes_since(Some(room_id), sender_user, *roomsince)
+					.changes_since(Some(room_id), sender_user, *roomsince, Some(next_batch))
 					.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
 					.collect()
 					.await,
@@ -644,7 +644,7 @@ async fn collect_account_data(
 
 	account_data.global = services
 		.account_data
-		.changes_since(None, sender_user, globalsince)
+		.changes_since(None, sender_user, globalsince, None)
 		.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
 		.collect()
 		.await;
@@ -655,7 +655,7 @@ async fn collect_account_data(
 				room.clone(),
 				services
 					.account_data
-					.changes_since(Some(room), sender_user, globalsince)
+					.changes_since(Some(room), sender_user, globalsince, None)
 					.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
 					.collect()
 					.await,
@@ -876,7 +876,7 @@ async fn collect_to_device(
 		next_batch: next_batch.to_string(),
 		events: services
 			.users
-			.get_to_device_events(sender_user, sender_device)
+			.get_to_device_events(sender_user, sender_device, None, Some(next_batch))
 			.collect()
 			.await,
 	})

src/api/server/event_auth.rs

@@ -1,7 +1,7 @@
 use std::{borrow::Borrow, iter::once};
 
 use axum::extract::State;
-use conduwuit::{Error, Result};
+use conduwuit::{utils::stream::ReadyExt, Error, Result};
 use futures::StreamExt;
 use ruma::{
 	api::{client::error::ErrorKind, federation::authorization::get_event_authorization},
@@ -48,7 +48,7 @@ pub(crate) async fn get_event_authorization_route(
 		.rooms
 		.auth_chain
 		.event_ids_iter(room_id, once(body.event_id.borrow()))
-		.await?
+		.ready_filter_map(Result::ok)
 		.filter_map(|id| async move { services.rooms.timeline.get_pdu_json(&id).await.ok() })
 		.then(|pdu| services.sending.convert_to_outgoing_federation_event(pdu))
 		.collect()

src/api/server/send_join.rs

@@ -238,8 +238,6 @@ async fn create_join_event(
 		.rooms
 		.auth_chain
 		.event_ids_iter(room_id, starting_events)
-		.await?
-		.map(Ok)
 		.broad_and_then(|event_id| async move {
 			services.rooms.timeline.get_pdu_json(&event_id).await
 		})

src/api/server/state.rs

@@ -56,8 +56,6 @@ pub(crate) async fn get_room_state_route(
 		.rooms
 		.auth_chain
 		.event_ids_iter(&body.room_id, once(body.event_id.borrow()))
-		.await?
-		.map(Ok)
 		.and_then(|id| async move { services.rooms.timeline.get_pdu_json(&id).await })
 		.and_then(|pdu| {
 			services

src/api/server/state_ids.rs

@@ -2,7 +2,7 @@ use std::{borrow::Borrow, iter::once};
 
 use axum::extract::State;
 use conduwuit::{at, err, Result};
-use futures::StreamExt;
+use futures::{StreamExt, TryStreamExt};
 use ruma::{api::federation::event::get_room_state_ids, OwnedEventId};
 
 use super::AccessCheck;
@@ -44,10 +44,8 @@ pub(crate) async fn get_room_state_ids_route(
 		.rooms
 		.auth_chain
 		.event_ids_iter(&body.room_id, once(body.event_id.borrow()))
-		.await?
-		.map(|id| (*id).to_owned())
-		.collect()
-		.await;
+		.try_collect()
+		.await?;
 
 	Ok(get_room_state_ids::v1::Response { auth_chain_ids, pdu_ids })
 }

src/core/config/check.rs

@@ -38,7 +38,7 @@ pub fn check(config: &Config) -> Result {
 		));
 	}
 
-	if cfg!(all(feature = "hardened_malloc", feature = "jemalloc")) {
+	if cfg!(all(feature = "hardened_malloc", feature = "jemalloc", not(target_env = "msvc"))) {
 		debug_warn!(
 			"hardened_malloc and jemalloc compile-time features are both enabled, this causes \
 			 jemalloc to be used."

src/core/config/mod.rs

@@ -52,7 +52,7 @@ use crate::{err, error::Error, utils::sys, Result};
 ### For more information, see:
 ### https://conduwuit.puppyirl.gay/configuration.html
 "#,
-	ignore = "catchall well_known tls"
+	ignore = "catchall well_known tls blurhashing"
 )]
 pub struct Config {
 	/// The server_name is the pretty name of this server. It is used as a
@@ -480,6 +480,36 @@ pub struct Config {
 	#[serde(default = "default_pusher_idle_timeout")]
 	pub pusher_idle_timeout: u64,
 
+	/// Maximum time to receive a request from a client (seconds).
+	///
+	/// default: 75
+	#[serde(default = "default_client_receive_timeout")]
+	pub client_receive_timeout: u64,
+
+	/// Maximum time to process a request received from a client (seconds).
+	///
+	/// default: 180
+	#[serde(default = "default_client_request_timeout")]
+	pub client_request_timeout: u64,
+
+	/// Maximum time to transmit a response to a client (seconds)
+	///
+	/// default: 120
+	#[serde(default = "default_client_response_timeout")]
+	pub client_response_timeout: u64,
+
+	/// Grace period for clean shutdown of client requests (seconds).
+	///
+	/// default: 10
+	#[serde(default = "default_client_shutdown_timeout")]
+	pub client_shutdown_timeout: u64,
+
+	/// Grace period for clean shutdown of federation requests (seconds).
+	///
+	/// default: 5
+	#[serde(default = "default_sender_shutdown_timeout")]
+	pub sender_shutdown_timeout: u64,
+
 	/// Enables registration. If set to false, no users can register on this
 	/// server.
 	///
@@ -510,8 +540,9 @@ pub struct Config {
 	/// display: sensitive
 	pub registration_token: Option<String>,
 
-	/// Path to a file on the system that gets read for the registration token.
-	/// this config option takes precedence/priority over "registration_token".
+	/// Path to a file on the system that gets read for additional registration
+	/// tokens. Multiple tokens can be added if you separate them with
+	/// whitespace
 	///
 	/// conduwuit must be able to access the file, and it must not be empty
 	///
@@ -1758,6 +1789,9 @@ pub struct Config {
 	#[serde(default = "true_fn")]
 	pub config_reload_signal: bool,
 
+	// external structure; separate section
+	#[serde(default)]
+	pub blurhashing: BlurhashConfig,
 	#[serde(flatten)]
 	#[allow(clippy::zero_sized_map_values)]
 	// this is a catchall, the map shouldn't be zero at runtime
@@ -1808,6 +1842,31 @@ pub struct WellKnownConfig {
 	pub support_mxid: Option<OwnedUserId>,
 }
 
+#[derive(Clone, Copy, Debug, Deserialize, Default)]
+#[allow(rustdoc::broken_intra_doc_links, rustdoc::bare_urls)]
+#[config_example_generator(filename = "conduwuit-example.toml", section = "global.blurhashing")]
+pub struct BlurhashConfig {
+	/// blurhashing x component, 4 is recommended by https://blurha.sh/
+	///
+	/// default: 4
+	#[serde(default = "default_blurhash_x_component")]
+	pub components_x: u32,
+	/// blurhashing y component, 3 is recommended by https://blurha.sh/
+	///
+	/// default: 3
+	#[serde(default = "default_blurhash_y_component")]
+	pub components_y: u32,
+	/// Max raw size that the server will blurhash, this is the size of the
+	/// image after converting it to raw data, it should be higher than the
+	/// upload limit but not too high. The higher it is the higher the
+	/// potential load will be for clients requesting blurhashes. The default
+	/// is 33.55MB. Setting it to 0 disables blurhashing.
+	///
+	/// default: 33554432
+	#[serde(default = "default_blurhash_max_raw_size")]
+	pub blurhash_max_raw_size: u64,
+}
+
 #[derive(Deserialize, Clone, Debug)]
 #[serde(transparent)]
 struct ListeningPort {
@@ -2169,3 +2228,23 @@ fn default_stream_width_default() -> usize { 32 }
 fn default_stream_width_scale() -> f32 { 1.0 }
 
 fn default_stream_amplification() -> usize { 1024 }
+
+fn default_client_receive_timeout() -> u64 { 75 }
+
+fn default_client_request_timeout() -> u64 { 180 }
+
+fn default_client_response_timeout() -> u64 { 120 }
+
+fn default_client_shutdown_timeout() -> u64 { 15 }
+
+fn default_sender_shutdown_timeout() -> u64 { 5 }
+
+// blurhashing defaults recommended by https://blurha.sh/
+// 2^25
+pub(super) fn default_blurhash_max_raw_size() -> u64 { 33_554_432 }
+
+pub(super) fn default_blurhash_x_component() -> u32 { 4 }
+
+pub(super) fn default_blurhash_y_component() -> u32 { 3 }
+
+// end recommended & blurhashing defaults

src/core/debug.rs

@@ -1,6 +1,6 @@
 #![allow(clippy::disallowed_macros)]
 
-use std::{any::Any, panic};
+use std::{any::Any, env, panic, sync::LazyLock};
 
 // Export debug proc_macros
 pub use conduwuit_macros::recursion_depth;
@@ -58,16 +58,26 @@ pub const INFO_SPAN_LEVEL: Level = if cfg!(debug_assertions) {
 	Level::DEBUG
 };
 
-pub fn set_panic_trap() {
+pub static DEBUGGER: LazyLock<bool> =
+	LazyLock::new(|| env::var("_").unwrap_or_default().ends_with("gdb"));
+
+#[cfg_attr(debug_assertions, crate::ctor)]
+#[cfg_attr(not(debug_assertions), allow(dead_code))]
+fn set_panic_trap() {
+	if !*DEBUGGER {
+		return;
+	}
+
 	let next = panic::take_hook();
 	panic::set_hook(Box::new(move |info| {
 		panic_handler(info, &next);
 	}));
 }
 
-#[inline(always)]
+#[cold]
+#[inline(never)]
 #[allow(deprecated_in_future)]
-fn panic_handler(info: &panic::PanicHookInfo<'_>, next: &dyn Fn(&panic::PanicHookInfo<'_>)) {
+pub fn panic_handler(info: &panic::PanicHookInfo<'_>, next: &dyn Fn(&panic::PanicHookInfo<'_>)) {
 	trap();
 	next(info);
 }

src/core/log/console.rs

@@ -1,3 +1,5 @@
+use std::{env, io, sync::LazyLock};
+
 use tracing::{
 	field::{Field, Visit},
 	Event, Level, Subscriber,
@@ -7,12 +9,59 @@ use tracing_subscriber::{
 	fmt,
 	fmt::{
 		format::{Compact, DefaultVisitor, Format, Full, Pretty, Writer},
-		FmtContext, FormatEvent, FormatFields,
+		FmtContext, FormatEvent, FormatFields, MakeWriter,
 	},
 	registry::LookupSpan,
 };
 
-use crate::{Config, Result};
+use crate::{apply, Config, Result};
+
+static SYSTEMD_MODE: LazyLock<bool> =
+	LazyLock::new(|| env::var("SYSTEMD_EXEC_PID").is_ok() && env::var("JOURNAL_STREAM").is_ok());
+
+pub struct ConsoleWriter {
+	stdout: io::Stdout,
+	stderr: io::Stderr,
+	_journal_stream: [u64; 2],
+	use_stderr: bool,
+}
+
+impl ConsoleWriter {
+	#[must_use]
+	pub fn new(_config: &Config) -> Self {
+		let journal_stream = get_journal_stream();
+		Self {
+			stdout: io::stdout(),
+			stderr: io::stderr(),
+			_journal_stream: journal_stream.into(),
+			use_stderr: journal_stream.0 != 0,
+		}
+	}
+}
+
+impl<'a> MakeWriter<'a> for ConsoleWriter {
+	type Writer = &'a Self;
+
+	fn make_writer(&'a self) -> Self::Writer { self }
+}
+
+impl io::Write for &'_ ConsoleWriter {
+	fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+		if self.use_stderr {
+			self.stderr.lock().write(buf)
+		} else {
+			self.stdout.lock().write(buf)
+		}
+	}
+
+	fn flush(&mut self) -> io::Result<()> {
+		if self.use_stderr {
+			self.stderr.lock().flush()
+		} else {
+			self.stdout.lock().flush()
+		}
+	}
+}
 
 pub struct ConsoleFormat {
 	_compact: Format<Compact>,
@@ -20,10 +69,6 @@ pub struct ConsoleFormat {
 	pretty: Format<Pretty>,
 }
 
-struct ConsoleVisitor<'a> {
-	visitor: DefaultVisitor<'a>,
-}
-
 impl ConsoleFormat {
 	#[must_use]
 	pub fn new(config: &Config) -> Self {
@@ -68,6 +113,10 @@ where
 	}
 }
 
+struct ConsoleVisitor<'a> {
+	visitor: DefaultVisitor<'a>,
+}
+
 impl<'writer> FormatFields<'writer> for ConsoleFormat {
 	fn format_fields<R>(&self, writer: Writer<'writer>, fields: R) -> Result<(), std::fmt::Error>
 	where
@@ -92,3 +141,19 @@ impl Visit for ConsoleVisitor<'_> {
 		self.visitor.record_debug(field, value);
 	}
 }
+
+#[must_use]
+fn get_journal_stream() -> (u64, u64) {
+	is_systemd_mode()
+		.then(|| env::var("JOURNAL_STREAM").ok())
+		.flatten()
+		.as_deref()
+		.and_then(|s| s.split_once(':'))
+		.map(apply!(2, str::parse))
+		.map(apply!(2, Result::unwrap_or_default))
+		.unwrap_or((0, 0))
+}
+
+#[inline]
+#[must_use]
+pub fn is_systemd_mode() -> bool { *SYSTEMD_MODE }

src/core/log/mod.rs

@@ -2,14 +2,14 @@
 
 pub mod capture;
 pub mod color;
-mod console;
+pub mod console;
 pub mod fmt;
 pub mod fmt_span;
 mod reload;
 mod suppress;
 
 pub use capture::Capture;
-pub use console::ConsoleFormat;
+pub use console::{is_systemd_mode, ConsoleFormat, ConsoleWriter};
 pub use reload::{LogLevelReloadHandles, ReloadHandle};
 pub use suppress::Suppress;
 pub use tracing::Level;

src/core/metrics/mod.rs

@@ -19,8 +19,6 @@ pub struct Metrics {
 	runtime_intervals: std::sync::Mutex<Option<RuntimeIntervals>>,
 
 	// TODO: move stats
-	pub requests_spawn_active: AtomicU32,
-	pub requests_spawn_finished: AtomicU32,
 	pub requests_handle_active: AtomicU32,
 	pub requests_handle_finished: AtomicU32,
 	pub requests_panic: AtomicU32,
@@ -48,8 +46,6 @@ impl Metrics {
 			#[cfg(tokio_unstable)]
 			runtime_intervals: std::sync::Mutex::new(runtime_intervals),
 
-			requests_spawn_active: AtomicU32::new(0),
-			requests_spawn_finished: AtomicU32::new(0),
 			requests_handle_active: AtomicU32::new(0),
 			requests_handle_finished: AtomicU32::new(0),
 			requests_panic: AtomicU32::new(0),

src/core/pdu/redact.rs

@@ -90,3 +90,21 @@ pub fn copy_redacts(&self) -> (Option<OwnedEventId>, Box<RawJsonValue>) {
 
 	(self.redacts.clone(), self.content.clone())
 }
+
+#[implement(super::Pdu)]
+#[must_use]
+pub fn redacts_id(&self, room_version: &RoomVersionId) -> Option<OwnedEventId> {
+	use RoomVersionId::*;
+
+	if self.kind != TimelineEventType::RoomRedaction {
+		return None;
+	}
+
+	match *room_version {
+		| V1 | V2 | V3 | V4 | V5 | V6 | V7 | V8 | V9 | V10 => self.redacts.clone(),
+		| _ =>
+			self.get_content::<RoomRedactionEventContent>()
+				.ok()?
+				.redacts,
+	}
+}

src/core/server.rs

@@ -69,6 +69,10 @@ impl Server {
 			return Err!("Reloading not enabled");
 		}
 
+		#[cfg(all(feature = "systemd", target_os = "linux"))]
+		sd_notify::notify(true, &[sd_notify::NotifyState::Reloading])
+			.expect("failed to notify systemd of reloading state");
+
 		if self.reloading.swap(true, Ordering::AcqRel) {
 			return Err!("Reloading already in progress");
 		}
@@ -83,7 +87,7 @@ impl Server {
 		})
 	}
 
-	pub fn restart(&self) -> Result<()> {
+	pub fn restart(&self) -> Result {
 		if self.restarting.swap(true, Ordering::AcqRel) {
 			return Err!("Restart already in progress");
 		}
@@ -93,7 +97,11 @@ impl Server {
 		})
 	}
 
-	pub fn shutdown(&self) -> Result<()> {
+	pub fn shutdown(&self) -> Result {
+		#[cfg(all(feature = "systemd", target_os = "linux"))]
+		sd_notify::notify(true, &[sd_notify::NotifyState::Stopping])
+			.expect("failed to notify systemd of stopping state");
+
 		if self.stopping.swap(true, Ordering::AcqRel) {
 			return Err!("Shutdown already in progress");
 		}

src/core/utils/mod.rs

@@ -84,6 +84,17 @@ macro_rules! apply {
 	};
 }
 
+#[macro_export]
+macro_rules! pair_of {
+	($decl:ty) => {
+		($decl, $decl)
+	};
+
+	($init:expr) => {
+		($init, $init)
+	};
+}
+
 /// Functor for truthy
 #[macro_export]
 macro_rules! is_true {

src/database/de.rs

@@ -22,7 +22,7 @@ pub(crate) fn from_slice<'a, T>(buf: &'a [u8]) -> Result<T>
 where
 	T: Deserialize<'a>,
 {
-	let mut deserializer = Deserializer { buf, pos: 0, seq: false };
+	let mut deserializer = Deserializer { buf, pos: 0, rec: 0, seq: false };
 
 	T::deserialize(&mut deserializer).debug_inspect(|_| {
 		deserializer
@@ -35,6 +35,7 @@ where
 pub(crate) struct Deserializer<'de> {
 	buf: &'de [u8],
 	pos: usize,
+	rec: usize,
 	seq: bool,
 }
 
@@ -107,7 +108,7 @@ impl<'de> Deserializer<'de> {
 	/// consumed None is returned instead.
 	#[inline]
 	fn record_peek_byte(&self) -> Option<u8> {
-		let started = self.pos != 0;
+		let started = self.pos != 0 || self.rec > 0;
 		let buf = &self.buf[self.pos..];
 		debug_assert!(
 			!started || buf[0] == Self::SEP,
@@ -121,13 +122,14 @@ impl<'de> Deserializer<'de> {
 	/// the start of the next record. (Case for some sequences)
 	#[inline]
 	fn record_start(&mut self) {
-		let started = self.pos != 0;
+		let started = self.pos != 0 || self.rec > 0;
 		debug_assert!(
 			!started || self.buf[self.pos] == Self::SEP,
 			"Missing expected record separator at current position"
 		);
 
 		self.inc_pos(started.into());
+		self.inc_rec(1);
 	}
 
 	/// Consume all remaining bytes, which may include record separators,
@@ -157,6 +159,9 @@ impl<'de> Deserializer<'de> {
 		debug_assert!(self.pos <= self.buf.len(), "pos out of range");
 	}
 
+	#[inline]
+	fn inc_rec(&mut self, n: usize) { self.rec = self.rec.saturating_add(n); }
+
 	/// Unconsumed input bytes.
 	#[inline]
 	fn remaining(&self) -> Result<usize> {
@@ -270,8 +275,16 @@ impl<'a, 'de: 'a> de::Deserializer<'de> for &'a mut Deserializer<'de> {
 	}
 
 	#[cfg_attr(unabridged, tracing::instrument(level = "trace", skip_all))]
-	fn deserialize_option<V: Visitor<'de>>(self, _visitor: V) -> Result<V::Value> {
-		unhandled!("deserialize Option not implemented")
+	fn deserialize_option<V: Visitor<'de>>(self, visitor: V) -> Result<V::Value> {
+		if self
+			.buf
+			.get(self.pos)
+			.is_none_or(|b| *b == Deserializer::SEP)
+		{
+			visitor.visit_none()
+		} else {
+			visitor.visit_some(self)
+		}
 	}
 
 	#[cfg_attr(unabridged, tracing::instrument(level = "trace", skip_all))]

src/database/engine.rs

@@ -30,13 +30,13 @@ use crate::{
 };
 
 pub struct Engine {
+	pub(crate) db: Db,
+	pub(crate) pool: Arc<Pool>,
+	pub(crate) ctx: Arc<Context>,
 	pub(super) read_only: bool,
 	pub(super) secondary: bool,
 	pub(crate) checksums: bool,
 	corks: AtomicU32,
-	pub(crate) db: Db,
-	pub(crate) pool: Arc<Pool>,
-	pub(crate) ctx: Arc<Context>,
 }
 
 pub(crate) type Db = DBWithThreadMode<MultiThreaded>;

src/database/engine/files.rs

@@ -1,32 +1,15 @@
-use std::fmt::Write;
-
 use conduwuit::{implement, Result};
+use rocksdb::LiveFile as SstFile;
 
 use super::Engine;
+use crate::util::map_err;
 
 #[implement(Engine)]
-pub fn file_list(&self) -> Result<String> {
-	match self.db.live_files() {
-		| Err(e) => Ok(String::from(e)),
-		| Ok(mut files) => {
-			files.sort_by_key(|f| f.name.clone());
-			let mut res = String::new();
-			writeln!(res, "| lev  | sst  | keys | dels | size | column |")?;
-			writeln!(res, "| ---: | :--- | ---: | ---: | ---: | :---   |")?;
-			for file in files {
-				writeln!(
-					res,
-					"| {} | {:<13} | {:7}+ | {:4}- | {:9} | {} |",
-					file.level,
-					file.name,
-					file.num_entries,
-					file.num_deletions,
-					file.size,
-					file.column_family_name,
-				)?;
-			}
-
-			Ok(res)
-		},
-	}
+pub fn file_list(&self) -> impl Iterator<Item = Result<SstFile>> + Send {
+	self.db
+		.live_files()
+		.map_err(map_err)
+		.into_iter()
+		.flat_map(Vec::into_iter)
+		.map(Ok)
 }

src/database/engine/open.rs

@@ -56,13 +56,13 @@ pub(crate) async fn open(ctx: Arc<Context>, desc: &[Descriptor]) -> Result<Arc<S
 	);
 
 	Ok(Arc::new(Self {
+		db,
+		pool: ctx.pool.clone(),
+		ctx: ctx.clone(),
 		read_only: config.rocksdb_read_only,
 		secondary: config.rocksdb_secondary,
 		checksums: config.rocksdb_checksums,
 		corks: AtomicU32::new(0),
-		pool: ctx.pool.clone(),
-		db,
-		ctx,
 	}))
 }
 

src/database/map.rs

@@ -44,24 +44,24 @@ use crate::{watchers::Watchers, Engine};
 
 pub struct Map {
 	name: &'static str,
-	db: Arc<Engine>,
-	cf: Arc<ColumnFamily>,
 	watchers: Watchers,
-	write_options: WriteOptions,
+	cf: Arc<ColumnFamily>,
+	db: Arc<Engine>,
 	read_options: ReadOptions,
 	cache_read_options: ReadOptions,
+	write_options: WriteOptions,
 }
 
 impl Map {
 	pub(crate) fn open(db: &Arc<Engine>, name: &'static str) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
 			name,
-			db: db.clone(),
-			cf: open::open(db, name),
 			watchers: Watchers::default(),
-			write_options: write_options_default(db),
+			cf: open::open(db, name),
+			db: db.clone(),
 			read_options: read_options_default(db),
 			cache_read_options: cache_read_options_default(db),
+			write_options: write_options_default(db),
 		}))
 	}
 

src/database/map/open.rs

@@ -30,8 +30,5 @@ pub(super) fn open(db: &Arc<Engine>, name: &str) -> Arc<ColumnFamily> {
 	// lifetime parameter. We should not hold this handle, even in its Arc, after
 	// closing the database (dropping `Engine`). Since `Arc<Engine>` is a sibling
 	// member along with this handle in `Map`, that is prevented.
-	unsafe {
-		Arc::increment_strong_count(cf_ptr);
-		Arc::from_raw(cf_ptr)
-	}
+	unsafe { Arc::from_raw(cf_ptr) }
 }

src/database/pool.rs

@@ -13,7 +13,7 @@ use std::{
 use async_channel::{QueueStrategy, Receiver, RecvError, Sender};
 use conduwuit::{
 	debug, debug_warn, err, error, implement,
-	result::{DebugInspect, LogDebugErr},
+	result::DebugInspect,
 	trace,
 	utils::sys::compute::{get_affinity, nth_core_available, set_affinity},
 	Error, Result, Server,
@@ -290,9 +290,12 @@ fn worker_init(&self, id: usize) {
 	// affinity is empty (no-op) if there's only one queue
 	set_affinity(affinity.clone());
 
-	#[cfg(feature = "jemalloc")]
+	#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
 	if affinity.clone().count() == 1 && conduwuit::alloc::je::is_affine_arena() {
-		use conduwuit::alloc::je::this_thread::{arena_id, set_arena};
+		use conduwuit::{
+			alloc::je::this_thread::{arena_id, set_arena},
+			result::LogDebugErr,
+		};
 
 		let id = affinity.clone().next().expect("at least one id");
 

src/database/tests.rs

@@ -3,7 +3,7 @@
 use std::fmt::Debug;
 
 use arrayvec::ArrayVec;
-use conduwuit::ruma::{serde::Raw, RoomId, UserId};
+use conduwuit::ruma::{serde::Raw, EventId, RoomId, UserId};
 use serde::Serialize;
 
 use crate::{
@@ -389,3 +389,160 @@ fn de_complex() {
 
 	assert_eq!(arr, key, "deserialization of serialization does not match");
 }
+
+#[test]
+fn serde_tuple_option_value_some() {
+	let room_id: &RoomId = "!room:example.com".try_into().unwrap();
+	let user_id: &UserId = "@user:example.com".try_into().unwrap();
+
+	let mut aa = Vec::<u8>::new();
+	aa.extend_from_slice(room_id.as_bytes());
+	aa.push(0xFF);
+	aa.extend_from_slice(user_id.as_bytes());
+
+	let bb: (&RoomId, Option<&UserId>) = (room_id, Some(user_id));
+	let bbs = serialize_to_vec(&bb).expect("failed to serialize tuple");
+	assert_eq!(aa, bbs);
+
+	let cc: (&RoomId, Option<&UserId>) =
+		de::from_slice(&bbs).expect("failed to deserialize tuple");
+
+	assert_eq!(bb.1, cc.1);
+	assert_eq!(cc.0, bb.0);
+}
+
+#[test]
+fn serde_tuple_option_value_none() {
+	let room_id: &RoomId = "!room:example.com".try_into().unwrap();
+
+	let mut aa = Vec::<u8>::new();
+	aa.extend_from_slice(room_id.as_bytes());
+	aa.push(0xFF);
+
+	let bb: (&RoomId, Option<&UserId>) = (room_id, None);
+	let bbs = serialize_to_vec(&bb).expect("failed to serialize tuple");
+	assert_eq!(aa, bbs);
+
+	let cc: (&RoomId, Option<&UserId>) =
+		de::from_slice(&bbs).expect("failed to deserialize tuple");
+
+	assert_eq!(None, cc.1);
+	assert_eq!(cc.0, bb.0);
+}
+
+#[test]
+fn serde_tuple_option_none_value() {
+	let user_id: &UserId = "@user:example.com".try_into().unwrap();
+
+	let mut aa = Vec::<u8>::new();
+	aa.push(0xFF);
+	aa.extend_from_slice(user_id.as_bytes());
+
+	let bb: (Option<&RoomId>, &UserId) = (None, user_id);
+	let bbs = serialize_to_vec(&bb).expect("failed to serialize tuple");
+	assert_eq!(aa, bbs);
+
+	let cc: (Option<&RoomId>, &UserId) =
+		de::from_slice(&bbs).expect("failed to deserialize tuple");
+
+	assert_eq!(None, cc.0);
+	assert_eq!(cc.1, bb.1);
+}
+
+#[test]
+fn serde_tuple_option_some_value() {
+	let room_id: &RoomId = "!room:example.com".try_into().unwrap();
+	let user_id: &UserId = "@user:example.com".try_into().unwrap();
+
+	let mut aa = Vec::<u8>::new();
+	aa.extend_from_slice(room_id.as_bytes());
+	aa.push(0xFF);
+	aa.extend_from_slice(user_id.as_bytes());
+
+	let bb: (Option<&RoomId>, &UserId) = (Some(room_id), user_id);
+	let bbs = serialize_to_vec(&bb).expect("failed to serialize tuple");
+	assert_eq!(aa, bbs);
+
+	let cc: (Option<&RoomId>, &UserId) =
+		de::from_slice(&bbs).expect("failed to deserialize tuple");
+
+	assert_eq!(bb.0, cc.0);
+	assert_eq!(cc.1, bb.1);
+}
+
+#[test]
+fn serde_tuple_option_some_some() {
+	let room_id: &RoomId = "!room:example.com".try_into().unwrap();
+	let user_id: &UserId = "@user:example.com".try_into().unwrap();
+
+	let mut aa = Vec::<u8>::new();
+	aa.extend_from_slice(room_id.as_bytes());
+	aa.push(0xFF);
+	aa.extend_from_slice(user_id.as_bytes());
+
+	let bb: (Option<&RoomId>, Option<&UserId>) = (Some(room_id), Some(user_id));
+	let bbs = serialize_to_vec(&bb).expect("failed to serialize tuple");
+	assert_eq!(aa, bbs);
+
+	let cc: (Option<&RoomId>, Option<&UserId>) =
+		de::from_slice(&bbs).expect("failed to deserialize tuple");
+
+	assert_eq!(cc.0, bb.0);
+	assert_eq!(bb.1, cc.1);
+}
+
+#[test]
+fn serde_tuple_option_none_none() {
+	let aa = vec![0xFF];
+
+	let bb: (Option<&RoomId>, Option<&UserId>) = (None, None);
+	let bbs = serialize_to_vec(&bb).expect("failed to serialize tuple");
+	assert_eq!(aa, bbs);
+
+	let cc: (Option<&RoomId>, Option<&UserId>) =
+		de::from_slice(&bbs).expect("failed to deserialize tuple");
+
+	assert_eq!(cc.0, bb.0);
+	assert_eq!(None, cc.1);
+}
+
+#[test]
+fn serde_tuple_option_some_none_some() {
+	let room_id: &RoomId = "!room:example.com".try_into().unwrap();
+	let user_id: &UserId = "@user:example.com".try_into().unwrap();
+
+	let mut aa = Vec::<u8>::new();
+	aa.extend_from_slice(room_id.as_bytes());
+	aa.push(0xFF);
+	aa.push(0xFF);
+	aa.extend_from_slice(user_id.as_bytes());
+
+	let bb: (Option<&RoomId>, Option<&EventId>, Option<&UserId>) =
+		(Some(room_id), None, Some(user_id));
+
+	let bbs = serialize_to_vec(&bb).expect("failed to serialize tuple");
+	assert_eq!(aa, bbs);
+
+	let cc: (Option<&RoomId>, Option<&EventId>, Option<&UserId>) =
+		de::from_slice(&bbs).expect("failed to deserialize tuple");
+
+	assert_eq!(bb.0, cc.0);
+	assert_eq!(None, cc.1);
+	assert_eq!(bb.1, cc.1);
+	assert_eq!(bb.2, cc.2);
+}
+
+#[test]
+fn serde_tuple_option_none_none_none() {
+	let aa = vec![0xFF, 0xFF];
+
+	let bb: (Option<&RoomId>, Option<&EventId>, Option<&UserId>) = (None, None, None);
+	let bbs = serialize_to_vec(&bb).expect("failed to serialize tuple");
+	assert_eq!(aa, bbs);
+
+	let cc: (Option<&RoomId>, Option<&EventId>, Option<&UserId>) =
+		de::from_slice(&bbs).expect("failed to deserialize tuple");
+
+	assert_eq!(None, cc.0);
+	assert_eq!(bb, cc);
+}

src/main/Cargo.toml

@@ -49,6 +49,9 @@ default = [
 	"zstd_compression",
 ]
 
+blurhashing = [
+	"conduwuit-service/blurhashing",
+]
 brotli_compression = [
 	"conduwuit-api/brotli_compression",
 	"conduwuit-core/brotli_compression",

src/main/logging.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use conduwuit::{
 	config::Config,
 	debug_warn, err,
-	log::{capture, fmt_span, ConsoleFormat, LogLevelReloadHandles},
+	log::{capture, fmt_span, ConsoleFormat, ConsoleWriter, LogLevelReloadHandles},
 	result::UnwrapOrErr,
 	Result,
 };
@@ -30,7 +30,7 @@ pub(crate) fn init(
 		.with_span_events(console_span_events)
 		.event_format(ConsoleFormat::new(config))
 		.fmt_fields(ConsoleFormat::new(config))
-		.map_writer(|w| w);
+		.with_writer(ConsoleWriter::new(config));
 
 	let (console_reload_filter, console_reload_handle) =
 		reload::Layer::new(console_filter.clone());

src/main/runtime.rs

@@ -8,13 +8,11 @@ use std::{
 	time::Duration,
 };
 
+#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
+use conduwuit::result::LogDebugErr;
 use conduwuit::{
 	is_true,
-	result::LogDebugErr,
-	utils::{
-		available_parallelism,
-		sys::compute::{nth_core_available, set_affinity},
-	},
+	utils::sys::compute::{nth_core_available, set_affinity},
 	Result,
 };
 use tokio::runtime::Builder;
@@ -25,6 +23,7 @@ const WORKER_NAME: &str = "conduwuit:worker";
 const WORKER_MIN: usize = 2;
 const WORKER_KEEPALIVE: u64 = 36;
 const MAX_BLOCKING_THREADS: usize = 1024;
+#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
 const DISABLE_MUZZY_THRESHOLD: usize = 4;
 
 static WORKER_AFFINITY: OnceLock<bool> = OnceLock::new();
@@ -122,7 +121,7 @@ fn set_worker_affinity() {
 	set_worker_mallctl(id);
 }
 
-#[cfg(feature = "jemalloc")]
+#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
 fn set_worker_mallctl(id: usize) {
 	use conduwuit::alloc::je::{
 		is_affine_arena,
@@ -137,13 +136,13 @@ fn set_worker_mallctl(id: usize) {
 		.get()
 		.expect("GC_MUZZY initialized by runtime::new()");
 
-	let muzzy_auto_disable = available_parallelism() >= DISABLE_MUZZY_THRESHOLD;
+	let muzzy_auto_disable = conduwuit::utils::available_parallelism() >= DISABLE_MUZZY_THRESHOLD;
 	if matches!(muzzy_option, Some(false) | None if muzzy_auto_disable) {
 		set_muzzy_decay(-1).log_debug_err().ok();
 	}
 }
 
-#[cfg(not(feature = "jemalloc"))]
+#[cfg(any(not(feature = "jemalloc"), target_env = "msvc"))]
 fn set_worker_mallctl(_: usize) {}
 
 #[tracing::instrument(
@@ -189,7 +188,7 @@ fn thread_park() {
 }
 
 fn gc_on_park() {
-	#[cfg(feature = "jemalloc")]
+	#[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))]
 	conduwuit::alloc::je::this_thread::decay()
 		.log_debug_err()
 		.ok();

src/router/layers.rs

@@ -5,7 +5,7 @@ use axum::{
 	Router,
 };
 use axum_client_ip::SecureClientIpSource;
-use conduwuit::{error, Result, Server};
+use conduwuit::{debug, error, Result, Server};
 use conduwuit_api::router::state::Guard;
 use conduwuit_service::Services;
 use http::{
@@ -18,6 +18,7 @@ use tower_http::{
 	cors::{self, CorsLayer},
 	sensitive_headers::SetSensitiveHeadersLayer,
 	set_header::SetResponseHeaderLayer,
+	timeout::{RequestBodyTimeoutLayer, ResponseBodyTimeoutLayer, TimeoutLayer},
 	trace::{DefaultOnFailure, DefaultOnRequest, DefaultOnResponse, TraceLayer},
 };
 use tracing::Level;
@@ -48,9 +49,9 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
 	))]
 	let layers = layers.layer(compression_layer(server));
 
+	let services_ = services.clone();
 	let layers = layers
 		.layer(SetSensitiveHeadersLayer::new([header::AUTHORIZATION]))
-		.layer(axum::middleware::from_fn_with_state(Arc::clone(services), request::spawn))
 		.layer(
 			TraceLayer::new_for_http()
 				.make_span_with(tracing_span::<_>)
@@ -60,6 +61,9 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
 		)
 		.layer(axum::middleware::from_fn_with_state(Arc::clone(services), request::handle))
 		.layer(SecureClientIpSource::ConnectInfo.into_extension())
+		.layer(ResponseBodyTimeoutLayer::new(Duration::from_secs(server.config.client_response_timeout)))
+		.layer(RequestBodyTimeoutLayer::new(Duration::from_secs(server.config.client_receive_timeout)))
+		.layer(TimeoutLayer::new(Duration::from_secs(server.config.client_request_timeout)))
 		.layer(SetResponseHeaderLayer::if_not_present(
 			HeaderName::from_static("origin-agent-cluster"), // https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Origin-Agent-Cluster
 			HeaderValue::from_static("?1"),
@@ -86,7 +90,7 @@ pub(crate) fn build(services: &Arc<Services>) -> Result<(Router, Guard)> {
 		))
 		.layer(cors_layer(server))
 		.layer(body_limit_layer(server))
-		.layer(CatchPanicLayer::custom(catch_panic));
+		.layer(CatchPanicLayer::custom(move |panic| catch_panic(panic, services_.clone())));
 
 	let (router, guard) = router::build(services);
 	Ok((router.layer(layers), guard))
@@ -164,15 +168,14 @@ fn body_limit_layer(server: &Server) -> DefaultBodyLimit {
 #[allow(clippy::needless_pass_by_value)]
 fn catch_panic(
 	err: Box<dyn Any + Send + 'static>,
+	services: Arc<Services>,
 ) -> http::Response<http_body_util::Full<bytes::Bytes>> {
-	//TODO: XXX
-	/*
-		conduwuit_service::services()
-			.server
-			.metrics
-			.requests_panic
-			.fetch_add(1, std::sync::atomic::Ordering::Release);
-	*/
+	services
+		.server
+		.metrics
+		.requests_panic
+		.fetch_add(1, std::sync::atomic::Ordering::Release);
+
 	let details = if let Some(s) = err.downcast_ref::<String>() {
 		s.clone()
 	} else if let Some(s) = err.downcast_ref::<&str>() {
@@ -196,20 +199,26 @@ fn catch_panic(
 }
 
 fn tracing_span<T>(request: &http::Request<T>) -> tracing::Span {
-	let path = request.extensions().get::<MatchedPath>().map_or_else(
-		|| {
-			request
-				.uri()
-				.path_and_query()
-				.expect("all requests have a path")
-				.as_str()
-		},
-		truncated_matched_path,
-	);
+	let path = request
+		.extensions()
+		.get::<MatchedPath>()
+		.map_or_else(|| request_path_str(request), truncated_matched_path);
 
-	let method = request.method();
+	tracing::span! {
+		parent: None,
+		debug::INFO_SPAN_LEVEL,
+		"router",
+		method = %request.method(),
+		%path,
+	}
+}
 
-	tracing::debug_span!(parent: None, "router", %method, %path)
+fn request_path_str<T>(request: &http::Request<T>) -> &str {
+	request
+		.uri()
+		.path_and_query()
+		.expect("all requests have a path")
+		.as_str()
 }
 
 fn truncated_matched_path(path: &MatchedPath) -> &str {

src/router/request.rs

@@ -1,4 +1,8 @@
-use std::sync::{atomic::Ordering, Arc};
+use std::{
+	fmt::Debug,
+	sync::{atomic::Ordering, Arc},
+	time::Duration,
+};
 
 use axum::{
 	extract::State,
@@ -6,82 +10,18 @@ use axum::{
 };
 use conduwuit::{debug, debug_error, debug_warn, err, error, trace, Result};
 use conduwuit_service::Services;
+use futures::FutureExt;
 use http::{Method, StatusCode, Uri};
+use tokio::time::sleep;
+use tracing::Span;
 
-#[tracing::instrument(
-	parent = None,
-	level = "trace",
-	skip_all,
-	fields(
-		handled = %services
-			.server
-			.metrics
-			.requests_spawn_finished
-			.fetch_add(1, Ordering::Relaxed),
-		active = %services
-			.server
-			.metrics
-			.requests_spawn_active
-			.fetch_add(1, Ordering::Relaxed),
-	)
-)]
-pub(crate) async fn spawn(
-	State(services): State<Arc<Services>>,
-	req: http::Request<axum::body::Body>,
-	next: axum::middleware::Next,
-) -> Result<Response, StatusCode> {
-	let server = &services.server;
-
-	#[cfg(debug_assertions)]
-	conduwuit::defer! {{
-		_ = server
-			.metrics
-			.requests_spawn_active
-			.fetch_sub(1, Ordering::Relaxed);
-	}};
-
-	if !server.running() {
-		debug_warn!("unavailable pending shutdown");
-		return Err(StatusCode::SERVICE_UNAVAILABLE);
-	}
-
-	let fut = next.run(req);
-	let task = server.runtime().spawn(fut);
-	task.await.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
-}
-
-#[tracing::instrument(
-	level = "debug",
-	skip_all,
-	fields(
-		handled = %services
-			.server
-			.metrics
-			.requests_handle_finished
-			.fetch_add(1, Ordering::Relaxed),
-		active = %services
-			.server
-			.metrics
-			.requests_handle_active
-			.fetch_add(1, Ordering::Relaxed),
-	)
-)]
+#[tracing::instrument(name = "request", level = "debug", skip_all)]
 pub(crate) async fn handle(
 	State(services): State<Arc<Services>>,
 	req: http::Request<axum::body::Body>,
 	next: axum::middleware::Next,
 ) -> Result<Response, StatusCode> {
-	let server = &services.server;
-
-	#[cfg(debug_assertions)]
-	conduwuit::defer! {{
-		_ = server
-			.metrics
-			.requests_handle_active
-			.fetch_sub(1, Ordering::Relaxed);
-	}};
-
-	if !server.running() {
+	if !services.server.running() {
 		debug_warn!(
 			method = %req.method(),
 			uri = %req.uri(),
@@ -93,14 +33,74 @@ pub(crate) async fn handle(
 
 	let uri = req.uri().clone();
 	let method = req.method().clone();
-	let result = next.run(req).await;
-	handle_result(&method, &uri, result)
+	let services_ = services.clone();
+	let parent = Span::current();
+	let task = services.server.runtime().spawn(async move {
+		tokio::select! {
+			response = execute(&services_, req, next, parent) => response,
+			response = services_.server.until_shutdown()
+				.then(|()| {
+					let timeout = services_.server.config.client_shutdown_timeout;
+					let timeout = Duration::from_secs(timeout);
+					sleep(timeout)
+				})
+				.map(|()| StatusCode::SERVICE_UNAVAILABLE)
+				.map(IntoResponse::into_response) => response,
+		}
+	});
+
+	task.await
+		.map_err(unhandled)
+		.and_then(move |result| handle_result(&method, &uri, result))
+}
+
+#[tracing::instrument(
+	name = "handle",
+	level = "debug",
+	parent = parent,
+	skip_all,
+	fields(
+		active = %services
+			.server
+			.metrics
+			.requests_handle_active
+			.fetch_add(1, Ordering::Relaxed),
+		handled = %services
+			.server
+			.metrics
+			.requests_handle_finished
+			.load(Ordering::Relaxed),
+	)
+)]
+async fn execute(
+	// we made a safety contract that Services will not go out of scope
+	// during the request; this ensures a reference is accounted for at
+	// the base frame of the task regardless of its detachment.
+	services: &Arc<Services>,
+	req: http::Request<axum::body::Body>,
+	next: axum::middleware::Next,
+	parent: Span,
+) -> Response {
+	#[cfg(debug_assertions)]
+	conduwuit::defer! {{
+		_ = services.server
+			.metrics
+			.requests_handle_finished
+			.fetch_add(1, Ordering::Relaxed);
+		_ = services.server
+			.metrics
+			.requests_handle_active
+			.fetch_sub(1, Ordering::Relaxed);
+	}};
+
+	next.run(req).await
 }
 
 fn handle_result(method: &Method, uri: &Uri, result: Response) -> Result<Response, StatusCode> {
 	let status = result.status();
 	let reason = status.canonical_reason().unwrap_or("Unknown Reason");
 	let code = status.as_u16();
+
 	if status.is_server_error() {
 		error!(method = ?method, uri = ?uri, "{code} {reason}");
 	} else if status.is_client_error() {
@@ -117,3 +117,10 @@ fn handle_result(method: &Method, uri: &Uri, result: Response) -> Result<Respons
 
 	Ok(result)
 }
+
+#[cold]
+fn unhandled<Error: Debug>(e: Error) -> StatusCode {
+	error!("unhandled error or panic during request: {e:?}");
+
+	StatusCode::INTERNAL_SERVER_ERROR
+}

src/router/run.rs

@@ -100,10 +100,6 @@ pub(crate) async fn stop(services: Arc<Services>) -> Result<()> {
 		);
 	}
 
-	#[cfg(all(feature = "systemd", target_os = "linux"))]
-	sd_notify::notify(true, &[sd_notify::NotifyState::Stopping])
-		.expect("failed to notify systemd of stopping state");
-
 	info!("Shutdown complete.");
 	Ok(())
 }
@@ -122,10 +118,10 @@ async fn handle_shutdown(server: Arc<Server>, tx: Sender<()>, handle: axum_serve
 		error!("failed sending shutdown transaction to channel: {e}");
 	}
 
-	let timeout = Duration::from_secs(36);
+	let timeout = server.config.client_shutdown_timeout;
+	let timeout = Duration::from_secs(timeout);
 	debug!(
 		?timeout,
-		spawn_active = ?server.metrics.requests_spawn_active.load(Ordering::Relaxed),
 		handle_active = ?server.metrics.requests_handle_active.load(Ordering::Relaxed),
 		"Notifying for graceful shutdown"
 	);

src/router/serve/plain.rs

@@ -24,27 +24,20 @@ pub(super) async fn serve(
 	info!("Listening on {addrs:?}");
 	while join_set.join_next().await.is_some() {}
 
-	let spawn_active = server.metrics.requests_spawn_active.load(Ordering::Relaxed);
 	let handle_active = server
 		.metrics
 		.requests_handle_active
 		.load(Ordering::Relaxed);
 	debug_info!(
-		spawn_finished = server
-			.metrics
-			.requests_spawn_finished
-			.load(Ordering::Relaxed),
 		handle_finished = server
 			.metrics
 			.requests_handle_finished
 			.load(Ordering::Relaxed),
 		panics = server.metrics.requests_panic.load(Ordering::Relaxed),
-		spawn_active,
 		handle_active,
 		"Stopped listening on {addrs:?}",
 	);
 
-	debug_assert!(spawn_active == 0, "active request tasks are not joined");
 	debug_assert!(handle_active == 0, "active request handles still pending");
 
 	Ok(())

src/router/serve/tls.rs

@@ -17,14 +17,13 @@ pub(super) async fn serve(
 	addrs: Vec<SocketAddr>,
 ) -> Result {
 	let tls = &server.config.tls;
-	let certs = tls
-		.certs
-		.as_ref()
-		.ok_or(err!(Config("tls.certs", "Missing required value in tls config section")))?;
+	let certs = tls.certs.as_ref().ok_or_else(|| {
+		err!(Config("tls.certs", "Missing required value in tls config section"))
+	})?;
 	let key = tls
 		.key
 		.as_ref()
-		.ok_or(err!(Config("tls.key", "Missing required value in tls config section")))?;
+		.ok_or_else(|| err!(Config("tls.key", "Missing required value in tls config section")))?;
 
 	// we use ring for ruma and hashing state, but aws-lc-rs is the new default.
 	// without this, TLS mode will panic.

src/router/serve/unix.rs

@@ -159,7 +159,12 @@ async fn fini(server: &Arc<Server>, listener: UnixListener, mut tasks: JoinSet<(
 	drop(listener);
 
 	debug!("Waiting for requests to finish...");
-	while server.metrics.requests_spawn_active.load(Ordering::Relaxed) > 0 {
+	while server
+		.metrics
+		.requests_handle_active
+		.load(Ordering::Relaxed)
+		.gt(&0)
+	{
 		tokio::select! {
 			task = tasks.join_next() => if task.is_none() { break; },
 			() = sleep(FINI_POLL_INTERVAL) => {},

src/service/Cargo.toml

@@ -44,6 +44,7 @@ url_preview = [
 zstd_compression = [
     "reqwest/zstd",
 ]
+blurhashing = ["dep:image","dep:blurhash"]
 
 [dependencies]
 arrayvec.workspace = true
@@ -82,6 +83,8 @@ tracing.workspace = true
 url.workspace = true
 webpage.workspace = true
 webpage.optional = true
+blurhash.workspace = true
+blurhash.optional = true
 
 [lints]
 workspace = true

src/service/account_data/mod.rs

@@ -5,7 +5,7 @@ use conduwuit::{
 	utils::{result::LogErr, stream::TryIgnore, ReadyExt},
 	Err, Result,
 };
-use database::{Deserialized, Handle, Interfix, Json, Map};
+use database::{Deserialized, Handle, Ignore, Json, Map};
 use futures::{Stream, StreamExt, TryFutureExt};
 use ruma::{
 	events::{
@@ -131,18 +131,20 @@ pub fn changes_since<'a>(
 	room_id: Option<&'a RoomId>,
 	user_id: &'a UserId,
 	since: u64,
+	to: Option<u64>,
 ) -> impl Stream<Item = AnyRawAccountDataEvent> + Send + 'a {
-	let prefix = (room_id, user_id, Interfix);
-	let prefix = database::serialize_key(prefix).expect("failed to serialize prefix");
+	type Key<'a> = (Option<&'a RoomId>, &'a UserId, u64, Ignore);
 
 	// Skip the data that's exactly at since, because we sent that last time
 	let first_possible = (room_id, user_id, since.saturating_add(1));
 
 	self.db
 		.roomuserdataid_accountdata
-		.stream_from_raw(&first_possible)
+		.stream_from(&first_possible)
 		.ignore_err()
-		.ready_take_while(move |(k, _)| k.starts_with(&prefix))
+		.ready_take_while(move |((room_id_, user_id_, count, _), _): &(Key<'_>, _)| {
+			room_id == *room_id_ && user_id == *user_id_ && to.is_none_or(|to| *count <= to)
+		})
 		.map(move |(_, v)| {
 			match room_id {
 				| Some(_) => serde_json::from_slice::<Raw<AnyRoomAccountDataEvent>>(v)

src/service/admin/console.rs

@@ -1,10 +1,11 @@
 #![cfg(feature = "console")]
+
 use std::{
 	collections::VecDeque,
 	sync::{Arc, Mutex},
 };
 
-use conduwuit::{debug, defer, error, log, Server};
+use conduwuit::{debug, defer, error, log, log::is_systemd_mode, Server};
 use futures::future::{AbortHandle, Abortable};
 use ruma::events::room::message::RoomMessageEventContent;
 use rustyline_async::{Readline, ReadlineError, ReadlineEvent};
@@ -123,7 +124,7 @@ impl Console {
 	}
 
 	async fn readline(self: &Arc<Self>) -> Result<ReadlineEvent, ReadlineError> {
-		let _suppression = log::Suppress::new(&self.server);
+		let _suppression = (!is_systemd_mode()).then(|| log::Suppress::new(&self.server));
 
 		let (mut readline, _writer) = Readline::new(PROMPT.to_owned())?;
 		let self_ = Arc::clone(self);

src/service/config/mod.rs

@@ -43,7 +43,15 @@ impl Deref for Service {
 #[implement(Service)]
 fn handle_reload(&self) -> Result {
 	if self.server.config.config_reload_signal {
+		#[cfg(all(feature = "systemd", target_os = "linux"))]
+		sd_notify::notify(true, &[sd_notify::NotifyState::Reloading])
+			.expect("failed to notify systemd of reloading state");
+
 		self.reload(iter::empty())?;
+
+		#[cfg(all(feature = "systemd", target_os = "linux"))]
+		sd_notify::notify(true, &[sd_notify::NotifyState::Ready])
+			.expect("failed to notify systemd of ready state");
 	}
 
 	Ok(())

src/service/globals/data.rs

@@ -69,9 +69,8 @@ impl Data {
 	}
 
 	#[inline]
-	pub fn bump_database_version(&self, new_version: u64) -> Result<()> {
+	pub fn bump_database_version(&self, new_version: u64) {
 		self.global.raw_put(b"version", new_version);
-		Ok(())
 	}
 
 	#[inline]
@@ -79,7 +78,4 @@ impl Data {
 
 	#[inline]
 	pub fn backup_list(&self) -> Result<String> { self.db.db.backup_list() }
-
-	#[inline]
-	pub fn file_list(&self) -> Result<String> { self.db.db.file_list() }
 }

src/service/globals/mod.rs

@@ -64,7 +64,7 @@ impl crate::Service for Service {
 			admin_alias: OwnedRoomAliasId::try_from(format!("#admins:{}", &args.server.name))
 				.expect("#admins:server_name is valid alias name"),
 			server_user: UserId::parse_with_server_name(
-				String::from("conduit"),
+				String::from("conduwuit"),
 				&args.server.name,
 			)
 			.expect("@conduit:server_name is valid"),

src/service/media/blurhash.rs

@@ -0,0 +1,179 @@
+#[cfg(feature = "blurhashing")]
+use conduwuit::config::BlurhashConfig as CoreBlurhashConfig;
+use conduwuit::{implement, Result};
+
+use super::Service;
+
+#[implement(Service)]
+#[cfg(not(feature = "blurhashing"))]
+pub fn create_blurhash(
+	&self,
+	_file: &[u8],
+	_content_type: Option<&str>,
+	_file_name: Option<&str>,
+) -> Result<Option<String>> {
+	conduwuit::debug_warn!("blurhashing on upload support was not compiled");
+
+	Ok(None)
+}
+
+#[implement(Service)]
+#[cfg(feature = "blurhashing")]
+pub fn create_blurhash(
+	&self,
+	file: &[u8],
+	content_type: Option<&str>,
+	file_name: Option<&str>,
+) -> Result<Option<String>> {
+	let config = BlurhashConfig::from(self.services.server.config.blurhashing);
+
+	// since 0 means disabled blurhashing, skipped blurhashing
+	if config.size_limit == 0 {
+		return Ok(None);
+	}
+
+	get_blurhash_from_request(file, content_type, file_name, config)
+		.map_err(|e| conduwuit::err!(debug_error!("blurhashing error: {e}")))
+		.map(Some)
+}
+
+/// Returns the blurhash or a blurhash error which implements Display.
+#[tracing::instrument(
+	name = "blurhash",
+	level = "debug",
+	skip(data),
+	fields(
+		bytes = data.len(),
+	),
+)]
+#[cfg(feature = "blurhashing")]
+fn get_blurhash_from_request(
+	data: &[u8],
+	mime: Option<&str>,
+	filename: Option<&str>,
+	config: BlurhashConfig,
+) -> Result<String, BlurhashingError> {
+	// Get format image is supposed to be in
+	let format = get_format_from_data_mime_and_filename(data, mime, filename)?;
+
+	// Get the image reader for said image format
+	let decoder = get_image_decoder_with_format_and_data(format, data)?;
+
+	// Check image size makes sense before unpacking whole image
+	if is_image_above_size_limit(&decoder, config) {
+		return Err(BlurhashingError::ImageTooLarge);
+	}
+
+	let image = image::DynamicImage::from_decoder(decoder)?;
+
+	blurhash_an_image(&image, config)
+}
+
+/// Gets the Image Format value from the data,mime, and filename
+/// It first checks if the mime is a valid image format
+/// Then it checks if the filename has a format, otherwise just guess based on
+/// the binary data Assumes that mime and filename extension won't be for a
+/// different file format than file.
+#[cfg(feature = "blurhashing")]
+fn get_format_from_data_mime_and_filename(
+	data: &[u8],
+	mime: Option<&str>,
+	filename: Option<&str>,
+) -> Result<image::ImageFormat, BlurhashingError> {
+	let extension = filename
+		.map(std::path::Path::new)
+		.and_then(std::path::Path::extension)
+		.map(std::ffi::OsStr::to_string_lossy);
+
+	mime.or(extension.as_deref())
+		.and_then(image::ImageFormat::from_mime_type)
+		.map_or_else(|| image::guess_format(data).map_err(Into::into), Ok)
+}
+
+#[cfg(feature = "blurhashing")]
+fn get_image_decoder_with_format_and_data(
+	image_format: image::ImageFormat,
+	data: &[u8],
+) -> Result<Box<dyn image::ImageDecoder + '_>, BlurhashingError> {
+	let mut image_reader = image::ImageReader::new(std::io::Cursor::new(data));
+	image_reader.set_format(image_format);
+	Ok(Box::new(image_reader.into_decoder()?))
+}
+
+#[cfg(feature = "blurhashing")]
+fn is_image_above_size_limit<T: image::ImageDecoder>(
+	decoder: &T,
+	blurhash_config: BlurhashConfig,
+) -> bool {
+	decoder.total_bytes() >= blurhash_config.size_limit
+}
+
+#[cfg(feature = "blurhashing")]
+#[tracing::instrument(name = "encode", level = "debug", skip_all)]
+#[inline]
+fn blurhash_an_image(
+	image: &image::DynamicImage,
+	blurhash_config: BlurhashConfig,
+) -> Result<String, BlurhashingError> {
+	Ok(blurhash::encode_image(
+		blurhash_config.components_x,
+		blurhash_config.components_y,
+		&image.to_rgba8(),
+	)?)
+}
+
+#[derive(Clone, Copy, Debug)]
+pub struct BlurhashConfig {
+	pub components_x: u32,
+	pub components_y: u32,
+
+	/// size limit in bytes
+	pub size_limit: u64,
+}
+
+#[cfg(feature = "blurhashing")]
+impl From<CoreBlurhashConfig> for BlurhashConfig {
+	fn from(value: CoreBlurhashConfig) -> Self {
+		Self {
+			components_x: value.components_x,
+			components_y: value.components_y,
+			size_limit: value.blurhash_max_raw_size,
+		}
+	}
+}
+
+#[derive(Debug)]
+#[cfg(feature = "blurhashing")]
+pub enum BlurhashingError {
+	HashingLibError(Box<dyn std::error::Error + Send>),
+	#[cfg(feature = "blurhashing")]
+	ImageError(Box<image::ImageError>),
+	ImageTooLarge,
+}
+
+#[cfg(feature = "blurhashing")]
+impl From<image::ImageError> for BlurhashingError {
+	fn from(value: image::ImageError) -> Self { Self::ImageError(Box::new(value)) }
+}
+
+#[cfg(feature = "blurhashing")]
+impl From<blurhash::Error> for BlurhashingError {
+	fn from(value: blurhash::Error) -> Self { Self::HashingLibError(Box::new(value)) }
+}
+
+#[cfg(feature = "blurhashing")]
+impl std::fmt::Display for BlurhashingError {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		write!(f, "Blurhash Error:")?;
+		match &self {
+			| Self::ImageTooLarge => write!(f, "Image was too large to blurhash")?,
+			| Self::HashingLibError(e) =>
+				write!(f, "There was an error with the blurhashing library => {e}")?,
+			#[cfg(feature = "blurhashing")]
+			| Self::ImageError(e) =>
+				write!(f, "There was an error with the image loading library => {e}")?,
+		};
+
+		Ok(())
+	}
+}

src/service/media/migrations.rs

@@ -13,7 +13,7 @@ use conduwuit::{
 	warn, Config, Result,
 };
 
-use crate::{migrations, Services};
+use crate::Services;
 
 /// Migrates a media directory from legacy base64 file names to sha2 file names.
 /// All errors are fatal. Upon success the database is keyed to not perform this
@@ -48,12 +48,6 @@ pub(crate) async fn migrate_sha256_media(services: &Services) -> Result<()> {
 		}
 	}
 
-	// Apply fix from when sha256_media was backward-incompat and bumped the schema
-	// version from 13 to 14. For users satisfying these conditions we can go back.
-	if services.globals.db.database_version().await == 14 && migrations::DATABASE_VERSION == 13 {
-		services.globals.db.bump_database_version(13)?;
-	}
-
 	db["global"].insert(b"feat_sha256_media", []);
 	info!("Finished applying sha256_media");
 	Ok(())

src/service/media/mod.rs

@@ -1,10 +1,10 @@
+pub mod blurhash;
 mod data;
 pub(super) mod migrations;
 mod preview;
 mod remote;
 mod tests;
 mod thumbnail;
-
 use std::{path::PathBuf, sync::Arc, time::SystemTime};
 
 use async_trait::async_trait;

src/service/migrations.rs

@@ -27,15 +27,7 @@ use crate::{media, Services};
 /// - If database is opened at lesser version we apply migrations up to this.
 ///   Note that named-feature migrations may also be performed when opening at
 ///   equal or lesser version. These are expected to be backward-compatible.
-pub(crate) const DATABASE_VERSION: u64 = 13;
-
-/// Conduit's database version.
-///
-/// Conduit bumped the database version to 16, but did not introduce any
-/// breaking changes. Their database migrations are extremely fragile and risky,
-/// and also do not really apply to us, so just to retain Conduit -> conduwuit
-/// compatibility we'll check for both versions.
-pub(crate) const CONDUIT_DATABASE_VERSION: u64 = 16;
+pub(crate) const DATABASE_VERSION: u64 = 17;
 
 pub(crate) async fn migrations(services: &Services) -> Result<()> {
 	let users_count = services.users.count().await;
@@ -63,10 +55,7 @@ pub(crate) async fn migrations(services: &Services) -> Result<()> {
 async fn fresh(services: &Services) -> Result<()> {
 	let db = &services.db;
 
-	services
-		.globals
-		.db
-		.bump_database_version(DATABASE_VERSION)?;
+	services.globals.db.bump_database_version(DATABASE_VERSION);
 
 	db["global"].insert(b"feat_sha256_media", []);
 	db["global"].insert(b"fix_bad_double_separator_in_state_cache", []);
@@ -130,6 +119,7 @@ async fn migrate(services: &Services) -> Result<()> {
 		.get(b"fix_referencedevents_missing_sep")
 		.await
 		.is_not_found()
+		|| services.globals.db.database_version().await < 17
 	{
 		fix_referencedevents_missing_sep(services).await?;
 	}
@@ -138,15 +128,19 @@ async fn migrate(services: &Services) -> Result<()> {
 		.get(b"fix_readreceiptid_readreceipt_duplicates")
 		.await
 		.is_not_found()
+		|| services.globals.db.database_version().await < 17
 	{
 		fix_readreceiptid_readreceipt_duplicates(services).await?;
 	}
 
-	let version_match = services.globals.db.database_version().await == DATABASE_VERSION
-		|| services.globals.db.database_version().await == CONDUIT_DATABASE_VERSION;
+	if services.globals.db.database_version().await < 17 {
+		services.globals.db.bump_database_version(17);
+		info!("Migration: Bumped database version to 17");
+	}
 
-	assert!(
-		version_match,
+	assert_eq!(
+		services.globals.db.database_version().await,
+		DATABASE_VERSION,
 		"Failed asserting local database version {} is equal to known latest conduwuit database \
 		 version {}",
 		services.globals.db.database_version().await,
@@ -290,7 +284,7 @@ async fn db_lt_12(services: &Services) -> Result<()> {
 			.await?;
 	}
 
-	services.globals.db.bump_database_version(12)?;
+	services.globals.db.bump_database_version(12);
 	info!("Migration: 11 -> 12 finished");
 	Ok(())
 }
@@ -335,7 +329,7 @@ async fn db_lt_13(services: &Services) -> Result<()> {
 			.await?;
 	}
 
-	services.globals.db.bump_database_version(13)?;
+	services.globals.db.bump_database_version(13);
 	info!("Migration: 12 -> 13 finished");
 	Ok(())
 }

src/service/rooms/auth_chain/mod.rs

@@ -4,6 +4,7 @@ use std::{
 	collections::{BTreeSet, HashSet, VecDeque},
 	fmt::Debug,
 	sync::Arc,
+	time::Instant,
 };
 
 use conduwuit::{
@@ -14,7 +15,7 @@ use conduwuit::{
 	},
 	validated, warn, Err, Result,
 };
-use futures::{Stream, StreamExt, TryFutureExt, TryStreamExt};
+use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt};
 use ruma::{EventId, OwnedEventId, RoomId};
 
 use self::data::Data;
@@ -30,6 +31,8 @@ struct Services {
 	timeline: Dep<rooms::timeline::Service>,
 }
 
+type Bucket<'a> = BTreeSet<(u64, &'a EventId)>;
+
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
@@ -45,42 +48,22 @@ impl crate::Service for Service {
 }
 
 #[implement(Service)]
-pub async fn event_ids_iter<'a, I>(
+pub fn event_ids_iter<'a, I>(
 	&'a self,
-	room_id: &RoomId,
+	room_id: &'a RoomId,
 	starting_events: I,
-) -> Result<impl Stream<Item = OwnedEventId> + Send + '_>
+) -> impl Stream<Item = Result<OwnedEventId>> + Send + 'a
 where
 	I: Iterator<Item = &'a EventId> + Clone + Debug + ExactSizeIterator + Send + 'a,
 {
-	let stream = self
-		.get_event_ids(room_id, starting_events)
-		.await?
-		.into_iter()
-		.stream();
-
-	Ok(stream)
-}
-
-#[implement(Service)]
-pub async fn get_event_ids<'a, I>(
-	&'a self,
-	room_id: &RoomId,
-	starting_events: I,
-) -> Result<Vec<OwnedEventId>>
-where
-	I: Iterator<Item = &'a EventId> + Clone + Debug + ExactSizeIterator + Send + 'a,
-{
-	let chain = self.get_auth_chain(room_id, starting_events).await?;
-	let event_ids = self
-		.services
-		.short
-		.multi_get_eventid_from_short(chain.into_iter().stream())
-		.ready_filter_map(Result::ok)
-		.collect()
-		.await;
-
-	Ok(event_ids)
+	self.get_auth_chain(room_id, starting_events)
+		.map_ok(|chain| {
+			self.services
+				.short
+				.multi_get_eventid_from_short(chain.into_iter().stream())
+				.ready_filter(Result::is_ok)
+		})
+		.try_flatten_stream()
 }
 
 #[implement(Service)]
@@ -94,9 +77,9 @@ where
 	I: Iterator<Item = &'a EventId> + Clone + Debug + ExactSizeIterator + Send + 'a,
 {
 	const NUM_BUCKETS: usize = 50; //TODO: change possible w/o disrupting db?
-	const BUCKET: BTreeSet<(u64, &EventId)> = BTreeSet::new();
+	const BUCKET: Bucket<'_> = BTreeSet::new();
 
-	let started = std::time::Instant::now();
+	let started = Instant::now();
 	let mut starting_ids = self
 		.services
 		.short
@@ -120,53 +103,7 @@ where
 	let full_auth_chain: Vec<ShortEventId> = buckets
 		.into_iter()
 		.try_stream()
-		.broad_and_then(|chunk| async move {
-			let chunk_key: Vec<ShortEventId> = chunk.iter().map(at!(0)).collect();
-
-			if chunk_key.is_empty() {
-				return Ok(Vec::new());
-			}
-
-			if let Ok(cached) = self.get_cached_eventid_authchain(&chunk_key).await {
-				return Ok(cached.to_vec());
-			}
-
-			let chunk_cache: Vec<_> = chunk
-				.into_iter()
-				.try_stream()
-				.broad_and_then(|(shortid, event_id)| async move {
-					if let Ok(cached) = self.get_cached_eventid_authchain(&[shortid]).await {
-						return Ok(cached.to_vec());
-					}
-
-					let auth_chain = self.get_auth_chain_inner(room_id, event_id).await?;
-					self.cache_auth_chain_vec(vec![shortid], auth_chain.as_slice());
-					debug!(
-						?event_id,
-						elapsed = ?started.elapsed(),
-						"Cache missed event"
-					);
-
-					Ok(auth_chain)
-				})
-				.try_collect()
-				.map_ok(|chunk_cache: Vec<_>| chunk_cache.into_iter().flatten().collect())
-				.map_ok(|mut chunk_cache: Vec<_>| {
-					chunk_cache.sort_unstable();
-					chunk_cache.dedup();
-					chunk_cache
-				})
-				.await?;
-
-			self.cache_auth_chain_vec(chunk_key, chunk_cache.as_slice());
-			debug!(
-				chunk_cache_length = ?chunk_cache.len(),
-				elapsed = ?started.elapsed(),
-				"Cache missed chunk",
-			);
-
-			Ok(chunk_cache)
-		})
+		.broad_and_then(|chunk| self.get_auth_chain_outer(room_id, started, chunk))
 		.try_collect()
 		.map_ok(|auth_chain: Vec<_>| auth_chain.into_iter().flatten().collect())
 		.map_ok(|mut full_auth_chain: Vec<_>| {
@@ -174,6 +111,7 @@ where
 			full_auth_chain.dedup();
 			full_auth_chain
 		})
+		.boxed()
 		.await?;
 
 	debug!(
@@ -185,6 +123,60 @@ where
 	Ok(full_auth_chain)
 }
 
+#[implement(Service)]
+async fn get_auth_chain_outer(
+	&self,
+	room_id: &RoomId,
+	started: Instant,
+	chunk: Bucket<'_>,
+) -> Result<Vec<ShortEventId>> {
+	let chunk_key: Vec<ShortEventId> = chunk.iter().map(at!(0)).collect();
+
+	if chunk_key.is_empty() {
+		return Ok(Vec::new());
+	}
+
+	if let Ok(cached) = self.get_cached_eventid_authchain(&chunk_key).await {
+		return Ok(cached.to_vec());
+	}
+
+	let chunk_cache: Vec<_> = chunk
+		.into_iter()
+		.try_stream()
+		.broad_and_then(|(shortid, event_id)| async move {
+			if let Ok(cached) = self.get_cached_eventid_authchain(&[shortid]).await {
+				return Ok(cached.to_vec());
+			}
+
+			let auth_chain = self.get_auth_chain_inner(room_id, event_id).await?;
+			self.cache_auth_chain_vec(vec![shortid], auth_chain.as_slice());
+			debug!(
+				?event_id,
+				elapsed = ?started.elapsed(),
+				"Cache missed event"
+			);
+
+			Ok(auth_chain)
+		})
+		.try_collect()
+		.map_ok(|chunk_cache: Vec<_>| chunk_cache.into_iter().flatten().collect())
+		.map_ok(|mut chunk_cache: Vec<_>| {
+			chunk_cache.sort_unstable();
+			chunk_cache.dedup();
+			chunk_cache
+		})
+		.await?;
+
+	self.cache_auth_chain_vec(chunk_key, chunk_cache.as_slice());
+	debug!(
+		chunk_cache_length = ?chunk_cache.len(),
+		elapsed = ?started.elapsed(),
+		"Cache missed chunk",
+	);
+
+	Ok(chunk_cache)
+}
+
 #[implement(Service)]
 #[tracing::instrument(name = "inner", level = "trace", skip(self, room_id))]
 async fn get_auth_chain_inner(

src/service/rooms/event_handler/resolve_state.rs

@@ -5,17 +5,17 @@ use std::{
 };
 
 use conduwuit::{
-	debug, err, implement,
+	err, implement, trace,
 	utils::stream::{automatic_width, IterStream, ReadyExt, TryWidebandExt, WidebandExt},
-	Result,
+	Error, Result,
 };
-use futures::{FutureExt, StreamExt, TryStreamExt};
+use futures::{future::try_join, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
 use ruma::{
 	state_res::{self, StateMap},
 	OwnedEventId, RoomId, RoomVersionId,
 };
 
-use crate::rooms::state_compressor::CompressedStateEvent;
+use crate::rooms::state_compressor::CompressedState;
 
 #[implement(super::Service)]
 #[tracing::instrument(name = "resolve", level = "debug", skip_all)]
@@ -24,14 +24,14 @@ pub async fn resolve_state(
 	room_id: &RoomId,
 	room_version_id: &RoomVersionId,
 	incoming_state: HashMap<u64, OwnedEventId>,
-) -> Result<Arc<HashSet<CompressedStateEvent>>> {
-	debug!("Loading current room state ids");
+) -> Result<Arc<CompressedState>> {
+	trace!("Loading current room state ids");
 	let current_sstatehash = self
 		.services
 		.state
 		.get_room_shortstatehash(room_id)
-		.await
-		.map_err(|e| err!(Database(error!("No state for {room_id:?}: {e:?}"))))?;
+		.map_err(|e| err!(Database(error!("No state for {room_id:?}: {e:?}"))))
+		.await?;
 
 	let current_state_ids: HashMap<_, _> = self
 		.services
@@ -40,53 +40,44 @@ pub async fn resolve_state(
 		.collect()
 		.await;
 
+	trace!("Loading fork states");
 	let fork_states = [current_state_ids, incoming_state];
-	let auth_chain_sets: Vec<HashSet<OwnedEventId>> = fork_states
+	let auth_chain_sets = fork_states
 		.iter()
 		.try_stream()
-		.wide_and_then(|state| async move {
-			let starting_events = state.values().map(Borrow::borrow);
-
-			let auth_chain = self
-				.services
+		.wide_and_then(|state| {
+			self.services
 				.auth_chain
-				.get_event_ids(room_id, starting_events)
-				.await?
-				.into_iter()
-				.collect();
-
-			Ok(auth_chain)
+				.event_ids_iter(room_id, state.values().map(Borrow::borrow))
+				.try_collect()
 		})
-		.try_collect()
-		.await?;
+		.try_collect::<Vec<HashSet<OwnedEventId>>>();
 
-	debug!("Loading fork states");
-	let fork_states: Vec<StateMap<OwnedEventId>> = fork_states
-		.into_iter()
+	let fork_states = fork_states
+		.iter()
 		.stream()
-		.wide_then(|fork_state| async move {
+		.wide_then(|fork_state| {
 			let shortstatekeys = fork_state.keys().copied().stream();
-
-			let event_ids = fork_state.values().cloned().stream().boxed();
-
+			let event_ids = fork_state.values().cloned().stream();
 			self.services
 				.short
 				.multi_get_statekey_from_short(shortstatekeys)
 				.zip(event_ids)
 				.ready_filter_map(|(ty_sk, id)| Some((ty_sk.ok()?, id)))
 				.collect()
-				.await
 		})
-		.collect()
-		.await;
+		.map(Ok::<_, Error>)
+		.try_collect::<Vec<StateMap<OwnedEventId>>>();
 
-	debug!("Resolving state");
+	let (fork_states, auth_chain_sets) = try_join(fork_states, auth_chain_sets).await?;
+
+	trace!("Resolving state");
 	let state = self
-		.state_resolution(room_version_id, &fork_states, &auth_chain_sets)
+		.state_resolution(room_version_id, fork_states.iter(), &auth_chain_sets)
 		.boxed()
 		.await?;
 
-	debug!("State resolution done.");
+	trace!("State resolution done.");
 	let state_events: Vec<_> = state
 		.iter()
 		.stream()
@@ -99,8 +90,8 @@ pub async fn resolve_state(
 		.collect()
 		.await;
 
-	debug!("Compressing state...");
-	let new_room_state: HashSet<_> = self
+	trace!("Compressing state...");
+	let new_room_state: CompressedState = self
 		.services
 		.state_compressor
 		.compress_state_events(
@@ -116,20 +107,23 @@ pub async fn resolve_state(
 
 #[implement(super::Service)]
 #[tracing::instrument(name = "ruma", level = "debug", skip_all)]
-pub async fn state_resolution(
-	&self,
-	room_version: &RoomVersionId,
-	state_sets: &[StateMap<OwnedEventId>],
-	auth_chain_sets: &[HashSet<OwnedEventId>],
-) -> Result<StateMap<OwnedEventId>> {
+pub async fn state_resolution<'a, StateSets>(
+	&'a self,
+	room_version: &'a RoomVersionId,
+	state_sets: StateSets,
+	auth_chain_sets: &'a [HashSet<OwnedEventId>],
+) -> Result<StateMap<OwnedEventId>>
+where
+	StateSets: Iterator<Item = &'a StateMap<OwnedEventId>> + Clone + Send,
+{
 	state_res::resolve(
 		room_version,
-		state_sets.iter(),
+		state_sets,
 		auth_chain_sets,
 		&|event_id| self.event_fetch(event_id),
 		&|event_id| self.event_exists(event_id),
 		automatic_width(),
 	)
-	.await
 	.map_err(|e| err!(error!("State resolution failed: {e:?}")))
+	.await
 }

src/service/rooms/event_handler/state_at_incoming.rs

@@ -1,18 +1,20 @@
 use std::{
 	borrow::Borrow,
 	collections::{HashMap, HashSet},
+	iter::Iterator,
 	sync::Arc,
 };
 
 use conduwuit::{
-	debug, err, implement,
-	result::LogErr,
-	utils::stream::{BroadbandExt, IterStream},
+	debug, err, implement, trace,
+	utils::stream::{BroadbandExt, IterStream, ReadyExt, TryBroadbandExt, TryWidebandExt},
 	PduEvent, Result,
 };
-use futures::{FutureExt, StreamExt};
+use futures::{future::try_join, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
 use ruma::{state_res::StateMap, OwnedEventId, RoomId, RoomVersionId};
 
+use crate::rooms::short::ShortStateHash;
+
 // TODO: if we know the prev_events of the incoming event we can avoid the
 #[implement(super::Service)]
 // request and build the state from a known point and resolve if > 1 prev_event
@@ -70,87 +72,44 @@ pub(super) async fn state_at_incoming_resolved(
 	room_id: &RoomId,
 	room_version_id: &RoomVersionId,
 ) -> Result<Option<HashMap<u64, OwnedEventId>>> {
-	debug!("Calculating state at event using state res");
-	let mut extremity_sstatehashes = HashMap::with_capacity(incoming_pdu.prev_events.len());
-
-	let mut okay = true;
-	for prev_eventid in &incoming_pdu.prev_events {
-		let Ok(prev_event) = self.services.timeline.get_pdu(prev_eventid).await else {
-			okay = false;
-			break;
-		};
-
-		let Ok(sstatehash) = self
-			.services
-			.state_accessor
-			.pdu_shortstatehash(prev_eventid)
-			.await
-		else {
-			okay = false;
-			break;
-		};
-
-		extremity_sstatehashes.insert(sstatehash, prev_event);
-	}
-
-	if !okay {
+	trace!("Calculating extremity statehashes...");
+	let Ok(extremity_sstatehashes) = incoming_pdu
+		.prev_events
+		.iter()
+		.try_stream()
+		.broad_and_then(|prev_eventid| {
+			self.services
+				.timeline
+				.get_pdu(prev_eventid)
+				.map_ok(move |prev_event| (prev_eventid, prev_event))
+		})
+		.broad_and_then(|(prev_eventid, prev_event)| {
+			self.services
+				.state_accessor
+				.pdu_shortstatehash(prev_eventid)
+				.map_ok(move |sstatehash| (sstatehash, prev_event))
+		})
+		.try_collect::<HashMap<_, _>>()
+		.await
+	else {
 		return Ok(None);
-	}
+	};
 
-	let mut fork_states = Vec::with_capacity(extremity_sstatehashes.len());
-	let mut auth_chain_sets = Vec::with_capacity(extremity_sstatehashes.len());
-	for (sstatehash, prev_event) in extremity_sstatehashes {
-		let mut leaf_state: HashMap<_, _> = self
-			.services
-			.state_accessor
-			.state_full_ids(sstatehash)
-			.collect()
-			.await;
-
-		if let Some(state_key) = &prev_event.state_key {
-			let shortstatekey = self
-				.services
-				.short
-				.get_or_create_shortstatekey(&prev_event.kind.to_string().into(), state_key)
-				.await;
-
-			let event_id = &prev_event.event_id;
-			leaf_state.insert(shortstatekey, event_id.clone());
-			// Now it's the state after the pdu
-		}
-
-		let mut state = StateMap::with_capacity(leaf_state.len());
-		let mut starting_events = Vec::with_capacity(leaf_state.len());
-		for (k, id) in &leaf_state {
-			if let Ok((ty, st_key)) = self
-				.services
-				.short
-				.get_statekey_from_short(*k)
-				.await
-				.log_err()
-			{
-				// FIXME: Undo .to_string().into() when StateMap
-				//        is updated to use StateEventType
-				state.insert((ty.to_string().into(), st_key), id.clone());
-			}
-
-			starting_events.push(id.borrow());
-		}
-
-		let auth_chain: HashSet<OwnedEventId> = self
-			.services
-			.auth_chain
-			.get_event_ids(room_id, starting_events.into_iter())
-			.await?
+	trace!("Calculating fork states...");
+	let (fork_states, auth_chain_sets): (Vec<StateMap<_>>, Vec<HashSet<_>>) =
+		extremity_sstatehashes
 			.into_iter()
-			.collect();
-
-		auth_chain_sets.push(auth_chain);
-		fork_states.push(state);
-	}
+			.try_stream()
+			.wide_and_then(|(sstatehash, prev_event)| {
+				self.state_at_incoming_fork(room_id, sstatehash, prev_event)
+			})
+			.try_collect()
+			.map_ok(Vec::into_iter)
+			.map_ok(Iterator::unzip)
+			.await?;
 
 	let Ok(new_state) = self
-		.state_resolution(room_version_id, &fork_states, &auth_chain_sets)
+		.state_resolution(room_version_id, fork_states.iter(), &auth_chain_sets)
 		.boxed()
 		.await
 	else {
@@ -158,16 +117,65 @@ pub(super) async fn state_at_incoming_resolved(
 	};
 
 	new_state
-		.iter()
+		.into_iter()
 		.stream()
-		.broad_then(|((event_type, state_key), event_id)| {
+		.broad_then(|((event_type, state_key), event_id)| async move {
 			self.services
 				.short
-				.get_or_create_shortstatekey(event_type, state_key)
-				.map(move |shortstatekey| (shortstatekey, event_id.clone()))
+				.get_or_create_shortstatekey(&event_type, &state_key)
+				.map(move |shortstatekey| (shortstatekey, event_id))
+				.await
 		})
 		.collect()
 		.map(Some)
 		.map(Ok)
 		.await
 }
+
+#[implement(super::Service)]
+async fn state_at_incoming_fork(
+	&self,
+	room_id: &RoomId,
+	sstatehash: ShortStateHash,
+	prev_event: PduEvent,
+) -> Result<(StateMap<OwnedEventId>, HashSet<OwnedEventId>)> {
+	let mut leaf_state: HashMap<_, _> = self
+		.services
+		.state_accessor
+		.state_full_ids(sstatehash)
+		.collect()
+		.await;
+
+	if let Some(state_key) = &prev_event.state_key {
+		let shortstatekey = self
+			.services
+			.short
+			.get_or_create_shortstatekey(&prev_event.kind.to_string().into(), state_key)
+			.await;
+
+		let event_id = &prev_event.event_id;
+		leaf_state.insert(shortstatekey, event_id.clone());
+		// Now it's the state after the pdu
+	}
+
+	let auth_chain = self
+		.services
+		.auth_chain
+		.event_ids_iter(room_id, leaf_state.values().map(Borrow::borrow))
+		.try_collect();
+
+	let fork_state = leaf_state
+		.iter()
+		.stream()
+		.broad_then(|(k, id)| {
+			self.services
+				.short
+				.get_statekey_from_short(*k)
+				.map_ok(|(ty, sk)| ((ty, sk), id.clone()))
+		})
+		.ready_filter_map(Result::ok)
+		.collect()
+		.map(Ok);
+
+	try_join(fork_state, auth_chain).await
+}

src/service/rooms/event_handler/upgrade_outlier_pdu.rs

@@ -1,21 +1,22 @@
-use std::{
-	borrow::Borrow,
-	collections::{BTreeMap, HashSet},
-	sync::Arc,
-	time::Instant,
-};
+use std::{borrow::Borrow, collections::BTreeMap, iter::once, sync::Arc, time::Instant};
 
-use conduwuit::{debug, debug_info, err, implement, trace, warn, Err, Error, PduEvent, Result};
-use futures::{future::ready, StreamExt};
+use conduwuit::{
+	debug, debug_info, err, implement, trace,
+	utils::stream::{BroadbandExt, ReadyExt},
+	warn, Err, PduEvent, Result,
+};
+use futures::{future::ready, FutureExt, StreamExt};
 use ruma::{
-	api::client::error::ErrorKind,
-	events::{room::redaction::RoomRedactionEventContent, StateEventType, TimelineEventType},
+	events::StateEventType,
 	state_res::{self, EventTypeExt},
-	CanonicalJsonValue, RoomId, RoomVersionId, ServerName,
+	CanonicalJsonValue, RoomId, ServerName,
 };
 
 use super::{get_room_version_id, to_room_version};
-use crate::rooms::{state_compressor::HashSetCompressStateEvent, timeline::RawPduId};
+use crate::rooms::{
+	state_compressor::{CompressedState, HashSetCompressStateEvent},
+	timeline::RawPduId,
+};
 
 #[implement(super::Service)]
 pub(super) async fn upgrade_outlier_to_timeline_pdu(
@@ -123,46 +124,15 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 
 	// Soft fail check before doing state res
 	debug!("Performing soft-fail check");
-	let soft_fail = {
-		use RoomVersionId::*;
-
-		!auth_check
-			|| incoming_pdu.kind == TimelineEventType::RoomRedaction
-				&& match room_version_id {
-					| V1 | V2 | V3 | V4 | V5 | V6 | V7 | V8 | V9 | V10 => {
-						if let Some(redact_id) = &incoming_pdu.redacts {
-							!self
-								.services
-								.state_accessor
-								.user_can_redact(
-									redact_id,
-									&incoming_pdu.sender,
-									&incoming_pdu.room_id,
-									true,
-								)
-								.await?
-						} else {
-							false
-						}
-					},
-					| _ => {
-						let content: RoomRedactionEventContent = incoming_pdu.get_content()?;
-						if let Some(redact_id) = &content.redacts {
-							!self
-								.services
-								.state_accessor
-								.user_can_redact(
-									redact_id,
-									&incoming_pdu.sender,
-									&incoming_pdu.room_id,
-									true,
-								)
-								.await?
-						} else {
-							false
-						}
-					},
-				}
+	let soft_fail = match (auth_check, incoming_pdu.redacts_id(&room_version_id)) {
+		| (false, _) => true,
+		| (true, None) => false,
+		| (true, Some(redact_id)) =>
+			!self
+				.services
+				.state_accessor
+				.user_can_redact(&redact_id, &incoming_pdu.sender, &incoming_pdu.room_id, true)
+				.await?,
 	};
 
 	// 13. Use state resolution to find new room state
@@ -174,42 +144,34 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 	// Now we calculate the set of extremities this room has after the incoming
 	// event has been applied. We start with the previous extremities (aka leaves)
 	trace!("Calculating extremities");
-	let mut extremities: HashSet<_> = self
+	let extremities: Vec<_> = self
 		.services
 		.state
 		.get_forward_extremities(room_id)
 		.map(ToOwned::to_owned)
+		.ready_filter(|event_id| {
+			// Remove any that are referenced by this incoming event's prev_events
+			!incoming_pdu.prev_events.contains(event_id)
+		})
+		.broad_filter_map(|event_id| async move {
+			// Only keep those extremities were not referenced yet
+			self.services
+				.pdu_metadata
+				.is_event_referenced(room_id, &event_id)
+				.await
+				.eq(&false)
+				.then_some(event_id)
+		})
 		.collect()
 		.await;
 
-	// Remove any forward extremities that are referenced by this incoming event's
-	// prev_events
-	trace!(
-		"Calculated {} extremities; checking against {} prev_events",
+	debug!(
+		"Retained {} extremities checked against {} prev_events",
 		extremities.len(),
 		incoming_pdu.prev_events.len()
 	);
-	for prev_event in &incoming_pdu.prev_events {
-		extremities.remove(&(**prev_event));
-	}
 
-	// Only keep those extremities were not referenced yet
-	let mut retained = HashSet::new();
-	for id in &extremities {
-		if !self
-			.services
-			.pdu_metadata
-			.is_event_referenced(room_id, id)
-			.await
-		{
-			retained.insert(id.clone());
-		}
-	}
-
-	extremities.retain(|id| retained.contains(id));
-	debug!("Retained {} extremities. Compressing state", extremities.len());
-
-	let state_ids_compressed: HashSet<_> = self
+	let state_ids_compressed: Arc<CompressedState> = self
 		.services
 		.state_compressor
 		.compress_state_events(
@@ -218,10 +180,9 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 				.map(|(ssk, eid)| (ssk, eid.borrow())),
 		)
 		.collect()
+		.map(Arc::new)
 		.await;
 
-	let state_ids_compressed = Arc::new(state_ids_compressed);
-
 	if incoming_pdu.state_key.is_some() {
 		debug!("Event is a state-event. Deriving new room state");
 
@@ -260,12 +221,14 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 	//     if not soft fail it
 	if soft_fail {
 		debug!("Soft failing event");
+		let extremities = extremities.iter().map(Borrow::borrow);
+
 		self.services
 			.timeline
 			.append_incoming_pdu(
 				&incoming_pdu,
 				val,
-				extremities.iter().map(|e| (**e).to_owned()).collect(),
+				extremities,
 				state_ids_compressed,
 				soft_fail,
 				&state_lock,
@@ -273,27 +236,30 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu(
 			.await?;
 
 		// Soft fail, we keep the event as an outlier but don't add it to the timeline
-		warn!("Event was soft failed: {incoming_pdu:?}");
 		self.services
 			.pdu_metadata
 			.mark_event_soft_failed(&incoming_pdu.event_id);
 
-		return Err(Error::BadRequest(ErrorKind::InvalidParam, "Event has been soft failed"));
+		warn!("Event was soft failed: {incoming_pdu:?}");
+		return Err!(Request(InvalidParam("Event has been soft failed")));
 	}
 
-	trace!("Appending pdu to timeline");
-	extremities.insert(incoming_pdu.event_id.clone());
-
 	// Now that the event has passed all auth it is added into the timeline.
 	// We use the `state_at_event` instead of `state_after` so we accurately
 	// represent the state for this event.
+	trace!("Appending pdu to timeline");
+	let extremities = extremities
+		.iter()
+		.map(Borrow::borrow)
+		.chain(once(incoming_pdu.event_id.borrow()));
+
 	let pdu_id = self
 		.services
 		.timeline
 		.append_incoming_pdu(
 			&incoming_pdu,
 			val,
-			extremities.into_iter().collect(),
+			extremities,
 			state_ids_compressed,
 			soft_fail,
 			&state_lock,

src/service/rooms/spaces/mod.rs

@@ -582,7 +582,7 @@ impl Service {
 				parents.pop_front();
 				parents.push_back(room);
 
-				let short_room_ids: Vec<_> = parents
+				let next_short_room_ids: Vec<_> = parents
 					.iter()
 					.stream()
 					.filter_map(|room_id| async move {
@@ -591,16 +591,18 @@ impl Service {
 					.collect()
 					.await;
 
-				Some(
-					PaginationToken {
-						short_room_ids,
-						limit: UInt::new(max_depth)
-							.expect("When sent in request it must have been valid UInt"),
-						max_depth: UInt::new(max_depth)
-							.expect("When sent in request it must have been valid UInt"),
-						suggested_only,
-					}
-					.to_string(),
+				(next_short_room_ids != short_room_ids && !next_short_room_ids.is_empty()).then(
+					|| {
+						PaginationToken {
+							short_room_ids: next_short_room_ids,
+							limit: UInt::new(max_depth)
+								.expect("When sent in request it must have been valid UInt"),
+							max_depth: UInt::new(max_depth)
+								.expect("When sent in request it must have been valid UInt"),
+							suggested_only,
+						}
+						.to_string()
+					},
 				)
 			} else {
 				None

src/service/rooms/state/mod.rs

@@ -1,9 +1,4 @@
-use std::{
-	collections::{HashMap, HashSet},
-	fmt::Write,
-	iter::once,
-	sync::Arc,
-};
+use std::{collections::HashMap, fmt::Write, iter::once, sync::Arc};
 
 use conduwuit::{
 	err,
@@ -33,7 +28,7 @@ use crate::{
 	globals, rooms,
 	rooms::{
 		short::{ShortEventId, ShortStateHash},
-		state_compressor::{parse_compressed_state_event, CompressedStateEvent},
+		state_compressor::{parse_compressed_state_event, CompressedState},
 	},
 	Dep,
 };
@@ -102,10 +97,9 @@ impl Service {
 		&self,
 		room_id: &RoomId,
 		shortstatehash: u64,
-		statediffnew: Arc<HashSet<CompressedStateEvent>>,
-		_statediffremoved: Arc<HashSet<CompressedStateEvent>>,
-		state_lock: &RoomMutexGuard, /* Take mutex guard to make sure users get the room state
-		                              * mutex */
+		statediffnew: Arc<CompressedState>,
+		_statediffremoved: Arc<CompressedState>,
+		state_lock: &RoomMutexGuard,
 	) -> Result {
 		let event_ids = statediffnew
 			.iter()
@@ -176,7 +170,7 @@ impl Service {
 		&self,
 		event_id: &EventId,
 		room_id: &RoomId,
-		state_ids_compressed: Arc<HashSet<CompressedStateEvent>>,
+		state_ids_compressed: Arc<CompressedState>,
 	) -> Result<ShortStateHash> {
 		const KEY_LEN: usize = size_of::<ShortEventId>();
 		const VAL_LEN: usize = size_of::<ShortStateHash>();
@@ -209,12 +203,12 @@ impl Service {
 
 			let (statediffnew, statediffremoved) =
 				if let Some(parent_stateinfo) = states_parents.last() {
-					let statediffnew: HashSet<_> = state_ids_compressed
+					let statediffnew: CompressedState = state_ids_compressed
 						.difference(&parent_stateinfo.full_state)
 						.copied()
 						.collect();
 
-					let statediffremoved: HashSet<_> = parent_stateinfo
+					let statediffremoved: CompressedState = parent_stateinfo
 						.full_state
 						.difference(&state_ids_compressed)
 						.copied()
@@ -222,7 +216,7 @@ impl Service {
 
 					(Arc::new(statediffnew), Arc::new(statediffremoved))
 				} else {
-					(state_ids_compressed, Arc::new(HashSet::new()))
+					(state_ids_compressed, Arc::new(CompressedState::new()))
 				};
 			self.services.state_compressor.save_state_from_diff(
 				shortstatehash,
@@ -300,10 +294,10 @@ impl Service {
 			// TODO: statehash with deterministic inputs
 			let shortstatehash = self.services.globals.next_count()?;
 
-			let mut statediffnew = HashSet::new();
+			let mut statediffnew = CompressedState::new();
 			statediffnew.insert(new);
 
-			let mut statediffremoved = HashSet::new();
+			let mut statediffremoved = CompressedState::new();
 			if let Some(replaces) = replaces {
 				statediffremoved.insert(*replaces);
 			}
@@ -398,13 +392,14 @@ impl Service {
 			.ignore_err()
 	}
 
-	pub async fn set_forward_extremities(
-		&self,
-		room_id: &RoomId,
-		event_ids: Vec<OwnedEventId>,
-		_state_lock: &RoomMutexGuard, /* Take mutex guard to make sure users get the room
-		                               * state mutex */
-	) {
+	pub async fn set_forward_extremities<'a, I>(
+		&'a self,
+		room_id: &'a RoomId,
+		event_ids: I,
+		_state_lock: &'a RoomMutexGuard,
+	) where
+		I: Iterator<Item = &'a EventId> + Send + 'a,
+	{
 		let prefix = (room_id, Interfix);
 		self.db
 			.roomid_pduleaves
@@ -413,7 +408,7 @@ impl Service {
 			.ready_for_each(|key| self.db.roomid_pduleaves.remove(key))
 			.await;
 
-		for event_id in &event_ids {
+		for event_id in event_ids {
 			let key = (room_id, event_id);
 			self.db.roomid_pduleaves.put_raw(key, event_id);
 		}
@@ -428,60 +423,54 @@ impl Service {
 		sender: &UserId,
 		state_key: Option<&str>,
 		content: &serde_json::value::RawValue,
-	) -> Result<StateMap<Arc<PduEvent>>> {
+	) -> Result<StateMap<PduEvent>> {
 		let Ok(shortstatehash) = self.get_room_shortstatehash(room_id).await else {
 			return Ok(HashMap::new());
 		};
 
-		let mut sauthevents: HashMap<_, _> =
-			state_res::auth_types_for_event(kind, sender, state_key, content)?
-				.iter()
-				.stream()
-				.broad_filter_map(|(event_type, state_key)| {
-					self.services
-						.short
-						.get_shortstatekey(event_type, state_key)
-						.map_ok(move |ssk| (ssk, (event_type, state_key)))
-						.map(Result::ok)
-				})
-				.map(|(ssk, (event_type, state_key))| {
-					(ssk, (event_type.to_owned(), state_key.to_owned()))
-				})
-				.collect()
-				.await;
+		let auth_types = state_res::auth_types_for_event(kind, sender, state_key, content)?;
+
+		let sauthevents: HashMap<_, _> = auth_types
+			.iter()
+			.stream()
+			.broad_filter_map(|(event_type, state_key)| {
+				self.services
+					.short
+					.get_shortstatekey(event_type, state_key)
+					.map_ok(move |ssk| (ssk, (event_type, state_key)))
+					.map(Result::ok)
+			})
+			.collect()
+			.await;
 
 		let (state_keys, event_ids): (Vec<_>, Vec<_>) = self
 			.services
 			.state_accessor
 			.state_full_shortids(shortstatehash)
-			.await
-			.map_err(|e| err!(Database(error!(?room_id, ?shortstatehash, "{e:?}"))))?
-			.into_iter()
-			.filter_map(|(shortstatekey, shorteventid)| {
+			.ready_filter_map(Result::ok)
+			.ready_filter_map(|(shortstatekey, shorteventid)| {
 				sauthevents
-					.remove(&shortstatekey)
-					.map(|(event_type, state_key)| ((event_type, state_key), shorteventid))
+					.get(&shortstatekey)
+					.map(|(ty, sk)| ((ty, sk), shorteventid))
 			})
-			.unzip();
+			.unzip()
+			.await;
 
-		let auth_pdus = self
-			.services
+		self.services
 			.short
 			.multi_get_eventid_from_short(event_ids.into_iter().stream())
 			.zip(state_keys.into_iter().stream())
-			.ready_filter_map(|(event_id, tsk)| Some((tsk, event_id.ok()?)))
-			.broad_filter_map(|(tsk, event_id): (_, OwnedEventId)| async move {
+			.ready_filter_map(|(event_id, (ty, sk))| Some(((ty, sk), event_id.ok()?)))
+			.broad_filter_map(|((ty, sk), event_id): (_, OwnedEventId)| async move {
 				self.services
 					.timeline
 					.get_pdu(&event_id)
 					.await
-					.map(Arc::new)
-					.map(move |pdu| (tsk, pdu))
+					.map(move |pdu| (((*ty).clone(), (*sk).clone()), pdu))
 					.ok()
 			})
 			.collect()
-			.await;
-
-		Ok(auth_pdus)
+			.map(Ok)
+			.await
 	}
 }

src/service/rooms/state_accessor/mod.rs

@@ -1,22 +1,19 @@
+mod room_state;
+mod server_can;
+mod state;
+mod user_can;
+
 use std::{
-	borrow::Borrow,
 	fmt::Write,
 	sync::{Arc, Mutex as StdMutex, Mutex},
 };
 
 use conduwuit::{
-	at, err, error,
-	pdu::PduBuilder,
-	utils,
-	utils::{
-		math::{usize_from_f64, Expected},
-		stream::BroadbandExt,
-		IterStream, ReadyExt,
-	},
-	Err, Error, PduEvent, Result,
+	err, utils,
+	utils::math::{usize_from_f64, Expected},
+	Result,
 };
-use database::{Deserialized, Map};
-use futures::{FutureExt, Stream, StreamExt, TryFutureExt};
+use database::Map;
 use lru_cache::LruCache;
 use ruma::{
 	events::{
@@ -28,29 +25,19 @@ use ruma::{
 			guest_access::{GuestAccess, RoomGuestAccessEventContent},
 			history_visibility::{HistoryVisibility, RoomHistoryVisibilityEventContent},
 			join_rules::{AllowRule, JoinRule, RoomJoinRulesEventContent, RoomMembership},
-			member::{MembershipState, RoomMemberEventContent},
+			member::RoomMemberEventContent,
 			name::RoomNameEventContent,
-			power_levels::{RoomPowerLevels, RoomPowerLevelsEventContent},
 			topic::RoomTopicEventContent,
 		},
-		StateEventType, TimelineEventType,
+		StateEventType,
 	},
 	room::RoomType,
 	space::SpaceRoomJoinRule,
-	EventEncryptionAlgorithm, EventId, JsOption, OwnedEventId, OwnedRoomAliasId, OwnedRoomId,
-	OwnedServerName, OwnedUserId, RoomId, ServerName, UserId,
+	EventEncryptionAlgorithm, JsOption, OwnedRoomAliasId, OwnedRoomId, OwnedServerName,
+	OwnedUserId, RoomId, UserId,
 };
-use serde::Deserialize;
 
-use crate::{
-	rooms,
-	rooms::{
-		short::{ShortEventId, ShortStateHash, ShortStateKey},
-		state::RoomMutexGuard,
-		state_compressor::parse_compressed_state_event,
-	},
-	Dep,
-};
+use crate::{rooms, rooms::short::ShortStateHash, Dep};
 
 pub struct Service {
 	pub server_visibility_cache: Mutex<LruCache<(OwnedServerName, ShortStateHash), bool>>,
@@ -142,432 +129,6 @@ impl crate::Service for Service {
 }
 
 impl Service {
-	pub fn state_full(
-		&self,
-		shortstatehash: ShortStateHash,
-	) -> impl Stream<Item = ((StateEventType, String), PduEvent)> + Send + '_ {
-		self.state_full_pdus(shortstatehash)
-			.ready_filter_map(|pdu| {
-				Some(((pdu.kind.to_string().into(), pdu.state_key.clone()?), pdu))
-			})
-	}
-
-	pub fn state_full_pdus(
-		&self,
-		shortstatehash: ShortStateHash,
-	) -> impl Stream<Item = PduEvent> + Send + '_ {
-		let short_ids = self
-			.state_full_shortids(shortstatehash)
-			.map(|result| result.expect("missing shortstatehash"))
-			.map(Vec::into_iter)
-			.map(|iter| iter.map(at!(1)))
-			.map(IterStream::stream)
-			.flatten_stream()
-			.boxed();
-
-		self.services
-			.short
-			.multi_get_eventid_from_short(short_ids)
-			.ready_filter_map(Result::ok)
-			.broad_filter_map(move |event_id: OwnedEventId| async move {
-				self.services.timeline.get_pdu(&event_id).await.ok()
-			})
-	}
-
-	/// Builds a StateMap by iterating over all keys that start
-	/// with state_hash, this gives the full state for the given state_hash.
-	#[tracing::instrument(skip(self), level = "debug")]
-	pub fn state_full_ids<'a, Id>(
-		&'a self,
-		shortstatehash: ShortStateHash,
-	) -> impl Stream<Item = (ShortStateKey, Id)> + Send + 'a
-	where
-		Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a,
-		<Id as ToOwned>::Owned: Borrow<EventId>,
-	{
-		let shortids = self
-			.state_full_shortids(shortstatehash)
-			.map(|result| result.expect("missing shortstatehash"))
-			.map(|vec| vec.into_iter().unzip())
-			.boxed()
-			.shared();
-
-		let shortstatekeys = shortids
-			.clone()
-			.map(at!(0))
-			.map(Vec::into_iter)
-			.map(IterStream::stream)
-			.flatten_stream();
-
-		let shorteventids = shortids
-			.map(at!(1))
-			.map(Vec::into_iter)
-			.map(IterStream::stream)
-			.flatten_stream();
-
-		self.services
-			.short
-			.multi_get_eventid_from_short(shorteventids)
-			.zip(shortstatekeys)
-			.ready_filter_map(|(event_id, shortstatekey)| Some((shortstatekey, event_id.ok()?)))
-	}
-
-	/// Returns a single EventId from `room_id` with key (`event_type`,
-	/// `state_key`).
-	#[tracing::instrument(skip(self), level = "debug")]
-	pub async fn state_get_id<Id>(
-		&self,
-		shortstatehash: ShortStateHash,
-		event_type: &StateEventType,
-		state_key: &str,
-	) -> Result<Id>
-	where
-		Id: for<'de> Deserialize<'de> + Sized + ToOwned,
-		<Id as ToOwned>::Owned: Borrow<EventId>,
-	{
-		let shortstatekey = self
-			.services
-			.short
-			.get_shortstatekey(event_type, state_key)
-			.await?;
-
-		let full_state = self
-			.services
-			.state_compressor
-			.load_shortstatehash_info(shortstatehash)
-			.await
-			.map_err(|e| err!(Database(error!(?event_type, ?state_key, "Missing state: {e:?}"))))?
-			.pop()
-			.expect("there is always one layer")
-			.full_state;
-
-		let compressed = full_state
-			.iter()
-			.find(|bytes| bytes.starts_with(&shortstatekey.to_be_bytes()))
-			.ok_or(err!(Database("No shortstatekey in compressed state")))?;
-
-		let (_, shorteventid) = parse_compressed_state_event(*compressed);
-
-		self.services
-			.short
-			.get_eventid_from_short(shorteventid)
-			.await
-	}
-
-	#[inline]
-	pub async fn state_full_shortids(
-		&self,
-		shortstatehash: ShortStateHash,
-	) -> Result<Vec<(ShortStateKey, ShortEventId)>> {
-		let shortids = self
-			.services
-			.state_compressor
-			.load_shortstatehash_info(shortstatehash)
-			.await
-			.map_err(|e| err!(Database("Missing state IDs: {e}")))?
-			.pop()
-			.expect("there is always one layer")
-			.full_state
-			.iter()
-			.copied()
-			.map(parse_compressed_state_event)
-			.collect();
-
-		Ok(shortids)
-	}
-
-	/// Returns a single PDU from `room_id` with key (`event_type`,
-	/// `state_key`).
-	pub async fn state_get(
-		&self,
-		shortstatehash: ShortStateHash,
-		event_type: &StateEventType,
-		state_key: &str,
-	) -> Result<PduEvent> {
-		self.state_get_id(shortstatehash, event_type, state_key)
-			.and_then(|event_id: OwnedEventId| async move {
-				self.services.timeline.get_pdu(&event_id).await
-			})
-			.await
-	}
-
-	/// Returns a single PDU from `room_id` with key (`event_type`,`state_key`).
-	pub async fn state_get_content<T>(
-		&self,
-		shortstatehash: ShortStateHash,
-		event_type: &StateEventType,
-		state_key: &str,
-	) -> Result<T>
-	where
-		T: for<'de> Deserialize<'de>,
-	{
-		self.state_get(shortstatehash, event_type, state_key)
-			.await
-			.and_then(|event| event.get_content())
-	}
-
-	/// Get membership for given user in state
-	async fn user_membership(
-		&self,
-		shortstatehash: ShortStateHash,
-		user_id: &UserId,
-	) -> MembershipState {
-		self.state_get_content(shortstatehash, &StateEventType::RoomMember, user_id.as_str())
-			.await
-			.map_or(MembershipState::Leave, |c: RoomMemberEventContent| c.membership)
-	}
-
-	/// The user was a joined member at this state (potentially in the past)
-	#[inline]
-	async fn user_was_joined(&self, shortstatehash: ShortStateHash, user_id: &UserId) -> bool {
-		self.user_membership(shortstatehash, user_id).await == MembershipState::Join
-	}
-
-	/// The user was an invited or joined room member at this state (potentially
-	/// in the past)
-	#[inline]
-	async fn user_was_invited(&self, shortstatehash: ShortStateHash, user_id: &UserId) -> bool {
-		let s = self.user_membership(shortstatehash, user_id).await;
-		s == MembershipState::Join || s == MembershipState::Invite
-	}
-
-	/// Whether a server is allowed to see an event through federation, based on
-	/// the room's history_visibility at that event's state.
-	#[tracing::instrument(skip_all, level = "trace")]
-	pub async fn server_can_see_event(
-		&self,
-		origin: &ServerName,
-		room_id: &RoomId,
-		event_id: &EventId,
-	) -> bool {
-		let Ok(shortstatehash) = self.pdu_shortstatehash(event_id).await else {
-			return true;
-		};
-
-		if let Some(visibility) = self
-			.server_visibility_cache
-			.lock()
-			.expect("locked")
-			.get_mut(&(origin.to_owned(), shortstatehash))
-		{
-			return *visibility;
-		}
-
-		let history_visibility = self
-			.state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "")
-			.await
-			.map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| {
-				c.history_visibility
-			});
-
-		let current_server_members = self
-			.services
-			.state_cache
-			.room_members(room_id)
-			.ready_filter(|member| member.server_name() == origin);
-
-		let visibility = match history_visibility {
-			| HistoryVisibility::WorldReadable | HistoryVisibility::Shared => true,
-			| HistoryVisibility::Invited => {
-				// Allow if any member on requesting server was AT LEAST invited, else deny
-				current_server_members
-					.any(|member| self.user_was_invited(shortstatehash, member))
-					.await
-			},
-			| HistoryVisibility::Joined => {
-				// Allow if any member on requested server was joined, else deny
-				current_server_members
-					.any(|member| self.user_was_joined(shortstatehash, member))
-					.await
-			},
-			| _ => {
-				error!("Unknown history visibility {history_visibility}");
-				false
-			},
-		};
-
-		self.server_visibility_cache
-			.lock()
-			.expect("locked")
-			.insert((origin.to_owned(), shortstatehash), visibility);
-
-		visibility
-	}
-
-	/// Whether a user is allowed to see an event, based on
-	/// the room's history_visibility at that event's state.
-	#[tracing::instrument(skip_all, level = "trace")]
-	pub async fn user_can_see_event(
-		&self,
-		user_id: &UserId,
-		room_id: &RoomId,
-		event_id: &EventId,
-	) -> bool {
-		let Ok(shortstatehash) = self.pdu_shortstatehash(event_id).await else {
-			return true;
-		};
-
-		if let Some(visibility) = self
-			.user_visibility_cache
-			.lock()
-			.expect("locked")
-			.get_mut(&(user_id.to_owned(), shortstatehash))
-		{
-			return *visibility;
-		}
-
-		let currently_member = self.services.state_cache.is_joined(user_id, room_id).await;
-
-		let history_visibility = self
-			.state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "")
-			.await
-			.map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| {
-				c.history_visibility
-			});
-
-		let visibility = match history_visibility {
-			| HistoryVisibility::WorldReadable => true,
-			| HistoryVisibility::Shared => currently_member,
-			| HistoryVisibility::Invited => {
-				// Allow if any member on requesting server was AT LEAST invited, else deny
-				self.user_was_invited(shortstatehash, user_id).await
-			},
-			| HistoryVisibility::Joined => {
-				// Allow if any member on requested server was joined, else deny
-				self.user_was_joined(shortstatehash, user_id).await
-			},
-			| _ => {
-				error!("Unknown history visibility {history_visibility}");
-				false
-			},
-		};
-
-		self.user_visibility_cache
-			.lock()
-			.expect("locked")
-			.insert((user_id.to_owned(), shortstatehash), visibility);
-
-		visibility
-	}
-
-	/// Whether a user is allowed to see an event, based on
-	/// the room's history_visibility at that event's state.
-	#[tracing::instrument(skip_all, level = "trace")]
-	pub async fn user_can_see_state_events(&self, user_id: &UserId, room_id: &RoomId) -> bool {
-		if self.services.state_cache.is_joined(user_id, room_id).await {
-			return true;
-		}
-
-		let history_visibility = self
-			.room_state_get_content(room_id, &StateEventType::RoomHistoryVisibility, "")
-			.await
-			.map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| {
-				c.history_visibility
-			});
-
-		match history_visibility {
-			| HistoryVisibility::Invited =>
-				self.services.state_cache.is_invited(user_id, room_id).await,
-			| HistoryVisibility::WorldReadable => true,
-			| _ => false,
-		}
-	}
-
-	/// Returns the state hash for this pdu.
-	pub async fn pdu_shortstatehash(&self, event_id: &EventId) -> Result<ShortStateHash> {
-		const BUFSIZE: usize = size_of::<ShortEventId>();
-
-		self.services
-			.short
-			.get_shorteventid(event_id)
-			.and_then(|shorteventid| {
-				self.db
-					.shorteventid_shortstatehash
-					.aqry::<BUFSIZE, _>(&shorteventid)
-			})
-			.await
-			.deserialized()
-	}
-
-	/// Returns the full room state.
-	#[tracing::instrument(skip(self), level = "debug")]
-	pub fn room_state_full<'a>(
-		&'a self,
-		room_id: &'a RoomId,
-	) -> impl Stream<Item = Result<((StateEventType, String), PduEvent)>> + Send + 'a {
-		self.services
-			.state
-			.get_room_shortstatehash(room_id)
-			.map_ok(|shortstatehash| self.state_full(shortstatehash).map(Ok))
-			.map_err(move |e| err!(Database("Missing state for {room_id:?}: {e:?}")))
-			.try_flatten_stream()
-	}
-
-	/// Returns the full room state pdus
-	#[tracing::instrument(skip(self), level = "debug")]
-	pub fn room_state_full_pdus<'a>(
-		&'a self,
-		room_id: &'a RoomId,
-	) -> impl Stream<Item = Result<PduEvent>> + Send + 'a {
-		self.services
-			.state
-			.get_room_shortstatehash(room_id)
-			.map_ok(|shortstatehash| self.state_full_pdus(shortstatehash).map(Ok))
-			.map_err(move |e| err!(Database("Missing state for {room_id:?}: {e:?}")))
-			.try_flatten_stream()
-	}
-
-	/// Returns a single EventId from `room_id` with key (`event_type`,
-	/// `state_key`).
-	#[tracing::instrument(skip(self), level = "debug")]
-	pub async fn room_state_get_id<Id>(
-		&self,
-		room_id: &RoomId,
-		event_type: &StateEventType,
-		state_key: &str,
-	) -> Result<Id>
-	where
-		Id: for<'de> Deserialize<'de> + Sized + ToOwned,
-		<Id as ToOwned>::Owned: Borrow<EventId>,
-	{
-		self.services
-			.state
-			.get_room_shortstatehash(room_id)
-			.and_then(|shortstatehash| self.state_get_id(shortstatehash, event_type, state_key))
-			.await
-	}
-
-	/// Returns a single PDU from `room_id` with key (`event_type`,
-	/// `state_key`).
-	#[tracing::instrument(skip(self), level = "debug")]
-	pub async fn room_state_get(
-		&self,
-		room_id: &RoomId,
-		event_type: &StateEventType,
-		state_key: &str,
-	) -> Result<PduEvent> {
-		self.services
-			.state
-			.get_room_shortstatehash(room_id)
-			.and_then(|shortstatehash| self.state_get(shortstatehash, event_type, state_key))
-			.await
-	}
-
-	/// Returns a single PDU from `room_id` with key (`event_type`,`state_key`).
-	pub async fn room_state_get_content<T>(
-		&self,
-		room_id: &RoomId,
-		event_type: &StateEventType,
-		state_key: &str,
-	) -> Result<T>
-	where
-		T: for<'de> Deserialize<'de>,
-	{
-		self.room_state_get(room_id, event_type, state_key)
-			.await
-			.and_then(|event| event.get_content())
-	}
-
 	pub async fn get_name(&self, room_id: &RoomId) -> Result<String> {
 		self.room_state_get_content(room_id, &StateEventType::RoomName, "")
 			.await
@@ -592,28 +153,6 @@ impl Service {
 			.await
 	}
 
-	pub async fn user_can_invite(
-		&self,
-		room_id: &RoomId,
-		sender: &UserId,
-		target_user: &UserId,
-		state_lock: &RoomMutexGuard,
-	) -> bool {
-		self.services
-			.timeline
-			.create_hash_and_sign_event(
-				PduBuilder::state(
-					target_user.into(),
-					&RoomMemberEventContent::new(MembershipState::Invite),
-				),
-				sender,
-				room_id,
-				state_lock,
-			)
-			.await
-			.is_ok()
-	}
-
 	/// Checks if guests are able to view room content without joining
 	pub async fn is_world_readable(&self, room_id: &RoomId) -> bool {
 		self.room_state_get_content(room_id, &StateEventType::RoomHistoryVisibility, "")
@@ -649,74 +188,6 @@ impl Service {
 			.map(|c: RoomTopicEventContent| c.topic)
 	}
 
-	/// Checks if a given user can redact a given event
-	///
-	/// If federation is true, it allows redaction events from any user of the
-	/// same server as the original event sender
-	pub async fn user_can_redact(
-		&self,
-		redacts: &EventId,
-		sender: &UserId,
-		room_id: &RoomId,
-		federation: bool,
-	) -> Result<bool> {
-		let redacting_event = self.services.timeline.get_pdu(redacts).await;
-
-		if redacting_event
-			.as_ref()
-			.is_ok_and(|pdu| pdu.kind == TimelineEventType::RoomCreate)
-		{
-			return Err!(Request(Forbidden("Redacting m.room.create is not safe, forbidding.")));
-		}
-
-		if redacting_event
-			.as_ref()
-			.is_ok_and(|pdu| pdu.kind == TimelineEventType::RoomServerAcl)
-		{
-			return Err!(Request(Forbidden(
-				"Redacting m.room.server_acl will result in the room being inaccessible for \
-				 everyone (empty allow key), forbidding."
-			)));
-		}
-
-		if let Ok(pl_event_content) = self
-			.room_state_get_content::<RoomPowerLevelsEventContent>(
-				room_id,
-				&StateEventType::RoomPowerLevels,
-				"",
-			)
-			.await
-		{
-			let pl_event: RoomPowerLevels = pl_event_content.into();
-			Ok(pl_event.user_can_redact_event_of_other(sender)
-				|| pl_event.user_can_redact_own_event(sender)
-					&& if let Ok(redacting_event) = redacting_event {
-						if federation {
-							redacting_event.sender.server_name() == sender.server_name()
-						} else {
-							redacting_event.sender == sender
-						}
-					} else {
-						false
-					})
-		} else {
-			// Falling back on m.room.create to judge power level
-			if let Ok(room_create) = self
-				.room_state_get(room_id, &StateEventType::RoomCreate, "")
-				.await
-			{
-				Ok(room_create.sender == sender
-					|| redacting_event
-						.as_ref()
-						.is_ok_and(|redacting_event| redacting_event.sender == sender))
-			} else {
-				Err(Error::bad_database(
-					"No m.room.power_levels or m.room.create events in database for room",
-				))
-			}
-		}
-	}
-
 	/// Returns the join rule (`SpaceRoomJoinRule`) for a given room
 	pub async fn get_join_rule(
 		&self,

src/service/rooms/state_accessor/room_state.rs

@@ -0,0 +1,90 @@
+use std::borrow::Borrow;
+
+use conduwuit::{err, implement, PduEvent, Result};
+use futures::{Stream, StreamExt, TryFutureExt};
+use ruma::{events::StateEventType, EventId, RoomId};
+use serde::Deserialize;
+
+/// Returns a single PDU from `room_id` with key (`event_type`,`state_key`).
+#[implement(super::Service)]
+pub async fn room_state_get_content<T>(
+	&self,
+	room_id: &RoomId,
+	event_type: &StateEventType,
+	state_key: &str,
+) -> Result<T>
+where
+	T: for<'de> Deserialize<'de>,
+{
+	self.room_state_get(room_id, event_type, state_key)
+		.await
+		.and_then(|event| event.get_content())
+}
+
+/// Returns the full room state.
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub fn room_state_full<'a>(
+	&'a self,
+	room_id: &'a RoomId,
+) -> impl Stream<Item = Result<((StateEventType, String), PduEvent)>> + Send + 'a {
+	self.services
+		.state
+		.get_room_shortstatehash(room_id)
+		.map_ok(|shortstatehash| self.state_full(shortstatehash).map(Ok))
+		.map_err(move |e| err!(Database("Missing state for {room_id:?}: {e:?}")))
+		.try_flatten_stream()
+}
+
+/// Returns the full room state pdus
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub fn room_state_full_pdus<'a>(
+	&'a self,
+	room_id: &'a RoomId,
+) -> impl Stream<Item = Result<PduEvent>> + Send + 'a {
+	self.services
+		.state
+		.get_room_shortstatehash(room_id)
+		.map_ok(|shortstatehash| self.state_full_pdus(shortstatehash).map(Ok))
+		.map_err(move |e| err!(Database("Missing state for {room_id:?}: {e:?}")))
+		.try_flatten_stream()
+}
+
+/// Returns a single EventId from `room_id` with key (`event_type`,
+/// `state_key`).
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub async fn room_state_get_id<Id>(
+	&self,
+	room_id: &RoomId,
+	event_type: &StateEventType,
+	state_key: &str,
+) -> Result<Id>
+where
+	Id: for<'de> Deserialize<'de> + Sized + ToOwned,
+	<Id as ToOwned>::Owned: Borrow<EventId>,
+{
+	self.services
+		.state
+		.get_room_shortstatehash(room_id)
+		.and_then(|shortstatehash| self.state_get_id(shortstatehash, event_type, state_key))
+		.await
+}
+
+/// Returns a single PDU from `room_id` with key (`event_type`,
+/// `state_key`).
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub async fn room_state_get(
+	&self,
+	room_id: &RoomId,
+	event_type: &StateEventType,
+	state_key: &str,
+) -> Result<PduEvent> {
+	self.services
+		.state
+		.get_room_shortstatehash(room_id)
+		.and_then(|shortstatehash| self.state_get(shortstatehash, event_type, state_key))
+		.await
+}

src/service/rooms/state_accessor/server_can.rs

@@ -0,0 +1,73 @@
+use conduwuit::{error, implement, utils::stream::ReadyExt};
+use futures::StreamExt;
+use ruma::{
+	events::{
+		room::history_visibility::{HistoryVisibility, RoomHistoryVisibilityEventContent},
+		StateEventType,
+	},
+	EventId, RoomId, ServerName,
+};
+
+/// Whether a server is allowed to see an event through federation, based on
+/// the room's history_visibility at that event's state.
+#[implement(super::Service)]
+#[tracing::instrument(skip_all, level = "trace")]
+pub async fn server_can_see_event(
+	&self,
+	origin: &ServerName,
+	room_id: &RoomId,
+	event_id: &EventId,
+) -> bool {
+	let Ok(shortstatehash) = self.pdu_shortstatehash(event_id).await else {
+		return true;
+	};
+
+	if let Some(visibility) = self
+		.server_visibility_cache
+		.lock()
+		.expect("locked")
+		.get_mut(&(origin.to_owned(), shortstatehash))
+	{
+		return *visibility;
+	}
+
+	let history_visibility = self
+		.state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "")
+		.await
+		.map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| {
+			c.history_visibility
+		});
+
+	let current_server_members = self
+		.services
+		.state_cache
+		.room_members(room_id)
+		.ready_filter(|member| member.server_name() == origin);
+
+	let visibility = match history_visibility {
+		| HistoryVisibility::WorldReadable | HistoryVisibility::Shared => true,
+		| HistoryVisibility::Invited => {
+			// Allow if any member on requesting server was AT LEAST invited, else deny
+			current_server_members
+				.any(|member| self.user_was_invited(shortstatehash, member))
+				.await
+		},
+		| HistoryVisibility::Joined => {
+			// Allow if any member on requested server was joined, else deny
+			current_server_members
+				.any(|member| self.user_was_joined(shortstatehash, member))
+				.await
+		},
+		| _ => {
+			error!("Unknown history visibility {history_visibility}");
+			false
+		},
+	};
+
+	self.server_visibility_cache
+		.lock()
+		.expect("locked")
+		.insert((origin.to_owned(), shortstatehash), visibility);
+
+	visibility
+}

src/service/rooms/state_accessor/state.rs

@@ -0,0 +1,320 @@
+use std::{borrow::Borrow, ops::Deref, sync::Arc};
+
+use conduwuit::{
+	at, err, implement, pair_of,
+	utils::{
+		result::FlatOk,
+		stream::{BroadbandExt, IterStream, ReadyExt, TryExpect},
+	},
+	PduEvent, Result,
+};
+use database::Deserialized;
+use futures::{future::try_join, FutureExt, Stream, StreamExt, TryFutureExt};
+use ruma::{
+	events::{
+		room::member::{MembershipState, RoomMemberEventContent},
+		StateEventType,
+	},
+	EventId, OwnedEventId, UserId,
+};
+use serde::Deserialize;
+
+use crate::rooms::{
+	short::{ShortEventId, ShortStateHash, ShortStateKey},
+	state_compressor::{compress_state_event, parse_compressed_state_event, CompressedState},
+};
+
+/// The user was a joined member at this state (potentially in the past)
+#[implement(super::Service)]
+#[inline]
+pub async fn user_was_joined(&self, shortstatehash: ShortStateHash, user_id: &UserId) -> bool {
+	self.user_membership(shortstatehash, user_id).await == MembershipState::Join
+}
+
+/// The user was an invited or joined room member at this state (potentially
+/// in the past)
+#[implement(super::Service)]
+#[inline]
+pub async fn user_was_invited(&self, shortstatehash: ShortStateHash, user_id: &UserId) -> bool {
+	let s = self.user_membership(shortstatehash, user_id).await;
+	s == MembershipState::Join || s == MembershipState::Invite
+}
+
+/// Get membership for given user in state
+#[implement(super::Service)]
+pub async fn user_membership(
+	&self,
+	shortstatehash: ShortStateHash,
+	user_id: &UserId,
+) -> MembershipState {
+	self.state_get_content(shortstatehash, &StateEventType::RoomMember, user_id.as_str())
+		.await
+		.map_or(MembershipState::Leave, |c: RoomMemberEventContent| c.membership)
+}
+
+/// Returns a single PDU from `room_id` with key (`event_type`,`state_key`).
+#[implement(super::Service)]
+pub async fn state_get_content<T>(
+	&self,
+	shortstatehash: ShortStateHash,
+	event_type: &StateEventType,
+	state_key: &str,
+) -> Result<T>
+where
+	T: for<'de> Deserialize<'de>,
+{
+	self.state_get(shortstatehash, event_type, state_key)
+		.await
+		.and_then(|event| event.get_content())
+}
+
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub async fn state_contains(
+	&self,
+	shortstatehash: ShortStateHash,
+	event_type: &StateEventType,
+	state_key: &str,
+) -> bool {
+	let Ok(shortstatekey) = self
+		.services
+		.short
+		.get_shortstatekey(event_type, state_key)
+		.await
+	else {
+		return false;
+	};
+
+	self.state_contains_shortstatekey(shortstatehash, shortstatekey)
+		.await
+}
+
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub async fn state_contains_shortstatekey(
+	&self,
+	shortstatehash: ShortStateHash,
+	shortstatekey: ShortStateKey,
+) -> bool {
+	let start = compress_state_event(shortstatekey, 0);
+	let end = compress_state_event(shortstatekey, u64::MAX);
+
+	self.load_full_state(shortstatehash)
+		.map_ok(|full_state| full_state.range(start..=end).next().copied())
+		.await
+		.flat_ok()
+		.is_some()
+}
+
+/// Returns a single PDU from `room_id` with key (`event_type`,
+/// `state_key`).
+#[implement(super::Service)]
+pub async fn state_get(
+	&self,
+	shortstatehash: ShortStateHash,
+	event_type: &StateEventType,
+	state_key: &str,
+) -> Result<PduEvent> {
+	self.state_get_id(shortstatehash, event_type, state_key)
+		.and_then(|event_id: OwnedEventId| async move {
+			self.services.timeline.get_pdu(&event_id).await
+		})
+		.await
+}
+
+/// Returns a single EventId from `room_id` with key (`event_type`,
+/// `state_key`).
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub async fn state_get_id<Id>(
+	&self,
+	shortstatehash: ShortStateHash,
+	event_type: &StateEventType,
+	state_key: &str,
+) -> Result<Id>
+where
+	Id: for<'de> Deserialize<'de> + Sized + ToOwned,
+	<Id as ToOwned>::Owned: Borrow<EventId>,
+{
+	let shorteventid = self
+		.state_get_shortid(shortstatehash, event_type, state_key)
+		.await?;
+
+	self.services
+		.short
+		.get_eventid_from_short(shorteventid)
+		.await
+}
+
+/// Returns a single EventId from `room_id` with key (`event_type`,
+/// `state_key`).
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub async fn state_get_shortid(
+	&self,
+	shortstatehash: ShortStateHash,
+	event_type: &StateEventType,
+	state_key: &str,
+) -> Result<ShortEventId> {
+	let shortstatekey = self
+		.services
+		.short
+		.get_shortstatekey(event_type, state_key)
+		.await?;
+
+	let start = compress_state_event(shortstatekey, 0);
+	let end = compress_state_event(shortstatekey, u64::MAX);
+	self.load_full_state(shortstatehash)
+		.map_ok(|full_state| {
+			full_state
+				.range(start..=end)
+				.next()
+				.copied()
+				.map(parse_compressed_state_event)
+				.map(at!(1))
+				.ok_or(err!(Request(NotFound("Not found in room state"))))
+		})
+		.await?
+}
+
+/// Returns the state events removed between the interval (present in .0 but
+/// not in .1)
+#[implement(super::Service)]
+#[inline]
+pub fn state_removed(
+	&self,
+	shortstatehash: pair_of!(ShortStateHash),
+) -> impl Stream<Item = (ShortStateKey, ShortEventId)> + Send + '_ {
+	self.state_added((shortstatehash.1, shortstatehash.0))
+}
+
+/// Returns the state events added between the interval (present in .1 but
+/// not in .0)
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub fn state_added<'a>(
+	&'a self,
+	shortstatehash: pair_of!(ShortStateHash),
+) -> impl Stream<Item = (ShortStateKey, ShortEventId)> + Send + 'a {
+	let a = self.load_full_state(shortstatehash.0);
+	let b = self.load_full_state(shortstatehash.1);
+	try_join(a, b)
+		.map_ok(|(a, b)| b.difference(&a).copied().collect::<Vec<_>>())
+		.map_ok(IterStream::try_stream)
+		.try_flatten_stream()
+		.expect_ok()
+		.map(parse_compressed_state_event)
+}
+
+#[implement(super::Service)]
+pub fn state_full(
+	&self,
+	shortstatehash: ShortStateHash,
+) -> impl Stream<Item = ((StateEventType, String), PduEvent)> + Send + '_ {
+	self.state_full_pdus(shortstatehash)
+		.ready_filter_map(|pdu| {
+			Some(((pdu.kind.to_string().into(), pdu.state_key.clone()?), pdu))
+		})
+}
+
+#[implement(super::Service)]
+pub fn state_full_pdus(
+	&self,
+	shortstatehash: ShortStateHash,
+) -> impl Stream<Item = PduEvent> + Send + '_ {
+	let short_ids = self
+		.state_full_shortids(shortstatehash)
+		.expect_ok()
+		.map(at!(1));
+
+	self.services
+		.short
+		.multi_get_eventid_from_short(short_ids)
+		.ready_filter_map(Result::ok)
+		.broad_filter_map(move |event_id: OwnedEventId| async move {
+			self.services.timeline.get_pdu(&event_id).await.ok()
+		})
+}
+
+/// Builds a StateMap by iterating over all keys that start
+/// with state_hash, this gives the full state for the given state_hash.
+#[implement(super::Service)]
+#[tracing::instrument(skip(self), level = "debug")]
+pub fn state_full_ids<'a, Id>(
+	&'a self,
+	shortstatehash: ShortStateHash,
+) -> impl Stream<Item = (ShortStateKey, Id)> + Send + 'a
+where
+	Id: for<'de> Deserialize<'de> + Send + Sized + ToOwned + 'a,
+	<Id as ToOwned>::Owned: Borrow<EventId>,
+{
+	let shortids = self
+		.state_full_shortids(shortstatehash)
+		.expect_ok()
+		.unzip()
+		.shared();
+
+	let shortstatekeys = shortids
+		.clone()
+		.map(at!(0))
+		.map(Vec::into_iter)
+		.map(IterStream::stream)
+		.flatten_stream();
+
+	let shorteventids = shortids
+		.map(at!(1))
+		.map(Vec::into_iter)
+		.map(IterStream::stream)
+		.flatten_stream();
+
+	self.services
+		.short
+		.multi_get_eventid_from_short(shorteventids)
+		.zip(shortstatekeys)
+		.ready_filter_map(|(event_id, shortstatekey)| Some((shortstatekey, event_id.ok()?)))
+}
+
+#[implement(super::Service)]
+pub fn state_full_shortids(
+	&self,
+	shortstatehash: ShortStateHash,
+) -> impl Stream<Item = Result<(ShortStateKey, ShortEventId)>> + Send + '_ {
+	self.load_full_state(shortstatehash)
+		.map_ok(|full_state| {
+			full_state
+				.deref()
+				.iter()
+				.copied()
+				.map(parse_compressed_state_event)
+				.collect()
+		})
+		.map_ok(|vec: Vec<_>| vec.into_iter().try_stream())
+		.try_flatten_stream()
+}
+
+#[implement(super::Service)]
+async fn load_full_state(&self, shortstatehash: ShortStateHash) -> Result<Arc<CompressedState>> {
+	self.services
+		.state_compressor
+		.load_shortstatehash_info(shortstatehash)
+		.map_err(|e| err!(Database("Missing state IDs: {e}")))
+		.map_ok(|vec| vec.last().expect("at least one layer").full_state.clone())
+		.await
+}
+
+/// Returns the state hash for this pdu.
+#[implement(super::Service)]
+pub async fn pdu_shortstatehash(&self, event_id: &EventId) -> Result<ShortStateHash> {
+	const BUFSIZE: usize = size_of::<ShortEventId>();
+
+	self.services
+		.short
+		.get_shorteventid(event_id)
+		.and_then(|shorteventid| {
+			self.db
+				.shorteventid_shortstatehash
+				.aqry::<BUFSIZE, _>(&shorteventid)
+		})
+		.await
+		.deserialized()
+}

src/service/rooms/state_accessor/user_can.rs

@@ -0,0 +1,187 @@
+use conduwuit::{error, implement, pdu::PduBuilder, Err, Error, Result};
+use ruma::{
+	events::{
+		room::{
+			history_visibility::{HistoryVisibility, RoomHistoryVisibilityEventContent},
+			member::{MembershipState, RoomMemberEventContent},
+			power_levels::{RoomPowerLevels, RoomPowerLevelsEventContent},
+		},
+		StateEventType, TimelineEventType,
+	},
+	EventId, RoomId, UserId,
+};
+
+use crate::rooms::state::RoomMutexGuard;
+
+/// Checks if a given user can redact a given event
+///
+/// If federation is true, it allows redaction events from any user of the
+/// same server as the original event sender
+#[implement(super::Service)]
+pub async fn user_can_redact(
+	&self,
+	redacts: &EventId,
+	sender: &UserId,
+	room_id: &RoomId,
+	federation: bool,
+) -> Result<bool> {
+	let redacting_event = self.services.timeline.get_pdu(redacts).await;
+
+	if redacting_event
+		.as_ref()
+		.is_ok_and(|pdu| pdu.kind == TimelineEventType::RoomCreate)
+	{
+		return Err!(Request(Forbidden("Redacting m.room.create is not safe, forbidding.")));
+	}
+
+	if redacting_event
+		.as_ref()
+		.is_ok_and(|pdu| pdu.kind == TimelineEventType::RoomServerAcl)
+	{
+		return Err!(Request(Forbidden(
+			"Redacting m.room.server_acl will result in the room being inaccessible for \
+			 everyone (empty allow key), forbidding."
+		)));
+	}
+
+	if let Ok(pl_event_content) = self
+		.room_state_get_content::<RoomPowerLevelsEventContent>(
+			room_id,
+			&StateEventType::RoomPowerLevels,
+			"",
+		)
+		.await
+	{
+		let pl_event: RoomPowerLevels = pl_event_content.into();
+		Ok(pl_event.user_can_redact_event_of_other(sender)
+			|| pl_event.user_can_redact_own_event(sender)
+				&& if let Ok(redacting_event) = redacting_event {
+					if federation {
+						redacting_event.sender.server_name() == sender.server_name()
+					} else {
+						redacting_event.sender == sender
+					}
+				} else {
+					false
+				})
+	} else {
+		// Falling back on m.room.create to judge power level
+		if let Ok(room_create) = self
+			.room_state_get(room_id, &StateEventType::RoomCreate, "")
+			.await
+		{
+			Ok(room_create.sender == sender
+				|| redacting_event
+					.as_ref()
+					.is_ok_and(|redacting_event| redacting_event.sender == sender))
+		} else {
+			Err(Error::bad_database(
+				"No m.room.power_levels or m.room.create events in database for room",
+			))
+		}
+	}
+}
+
+/// Whether a user is allowed to see an event, based on
+/// the room's history_visibility at that event's state.
+#[implement(super::Service)]
+#[tracing::instrument(skip_all, level = "trace")]
+pub async fn user_can_see_event(
+	&self,
+	user_id: &UserId,
+	room_id: &RoomId,
+	event_id: &EventId,
+) -> bool {
+	let Ok(shortstatehash) = self.pdu_shortstatehash(event_id).await else {
+		return true;
+	};
+
+	if let Some(visibility) = self
+		.user_visibility_cache
+		.lock()
+		.expect("locked")
+		.get_mut(&(user_id.to_owned(), shortstatehash))
+	{
+		return *visibility;
+	}
+
+	let currently_member = self.services.state_cache.is_joined(user_id, room_id).await;
+
+	let history_visibility = self
+		.state_get_content(shortstatehash, &StateEventType::RoomHistoryVisibility, "")
+		.await
+		.map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| {
+			c.history_visibility
+		});
+
+	let visibility = match history_visibility {
+		| HistoryVisibility::WorldReadable => true,
+		| HistoryVisibility::Shared => currently_member,
+		| HistoryVisibility::Invited => {
+			// Allow if any member on requesting server was AT LEAST invited, else deny
+			self.user_was_invited(shortstatehash, user_id).await
+		},
+		| HistoryVisibility::Joined => {
+			// Allow if any member on requested server was joined, else deny
+			self.user_was_joined(shortstatehash, user_id).await
+		},
+		| _ => {
+			error!("Unknown history visibility {history_visibility}");
+			false
+		},
+	};
+
+	self.user_visibility_cache
+		.lock()
+		.expect("locked")
+		.insert((user_id.to_owned(), shortstatehash), visibility);
+
+	visibility
+}
+
+/// Whether a user is allowed to see an event, based on
+/// the room's history_visibility at that event's state.
+#[implement(super::Service)]
+#[tracing::instrument(skip_all, level = "trace")]
+pub async fn user_can_see_state_events(&self, user_id: &UserId, room_id: &RoomId) -> bool {
+	if self.services.state_cache.is_joined(user_id, room_id).await {
+		return true;
+	}
+
+	let history_visibility = self
+		.room_state_get_content(room_id, &StateEventType::RoomHistoryVisibility, "")
+		.await
+		.map_or(HistoryVisibility::Shared, |c: RoomHistoryVisibilityEventContent| {
+			c.history_visibility
+		});
+
+	match history_visibility {
+		| HistoryVisibility::Invited =>
+			self.services.state_cache.is_invited(user_id, room_id).await,
+		| HistoryVisibility::WorldReadable => true,
+		| _ => false,
+	}
+}
+
+#[implement(super::Service)]
+pub async fn user_can_invite(
+	&self,
+	room_id: &RoomId,
+	sender: &UserId,
+	target_user: &UserId,
+	state_lock: &RoomMutexGuard,
+) -> bool {
+	self.services
+		.timeline
+		.create_hash_and_sign_event(
+			PduBuilder::state(
+				target_user.into(),
+				&RoomMemberEventContent::new(MembershipState::Invite),
+			),
+			sender,
+			room_id,
+			state_lock,
+		)
+		.await
+		.is_ok()
+}

src/service/rooms/state_compressor/mod.rs

@@ -1,5 +1,5 @@
 use std::{
-	collections::{HashMap, HashSet},
+	collections::{BTreeSet, HashMap},
 	fmt::{Debug, Write},
 	mem::size_of,
 	sync::{Arc, Mutex},
@@ -63,8 +63,8 @@ type StateInfoLruCache = LruCache<ShortStateHash, ShortStateInfoVec>;
 type ShortStateInfoVec = Vec<ShortStateInfo>;
 type ParentStatesVec = Vec<ShortStateInfo>;
 
-pub(crate) type CompressedState = HashSet<CompressedStateEvent>;
-pub(crate) type CompressedStateEvent = [u8; 2 * size_of::<ShortId>()];
+pub type CompressedState = BTreeSet<CompressedStateEvent>;
+pub type CompressedStateEvent = [u8; 2 * size_of::<ShortId>()];
 
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
@@ -249,8 +249,8 @@ impl Service {
 	pub fn save_state_from_diff(
 		&self,
 		shortstatehash: ShortStateHash,
-		statediffnew: Arc<HashSet<CompressedStateEvent>>,
-		statediffremoved: Arc<HashSet<CompressedStateEvent>>,
+		statediffnew: Arc<CompressedState>,
+		statediffremoved: Arc<CompressedState>,
 		diff_to_sibling: usize,
 		mut parent_states: ParentStatesVec,
 	) -> Result {
@@ -363,7 +363,7 @@ impl Service {
 	pub async fn save_state(
 		&self,
 		room_id: &RoomId,
-		new_state_ids_compressed: Arc<HashSet<CompressedStateEvent>>,
+		new_state_ids_compressed: Arc<CompressedState>,
 	) -> Result<HashSetCompressStateEvent> {
 		let previous_shortstatehash = self
 			.services
@@ -396,12 +396,12 @@ impl Service {
 
 		let (statediffnew, statediffremoved) =
 			if let Some(parent_stateinfo) = states_parents.last() {
-				let statediffnew: HashSet<_> = new_state_ids_compressed
+				let statediffnew: CompressedState = new_state_ids_compressed
 					.difference(&parent_stateinfo.full_state)
 					.copied()
 					.collect();
 
-				let statediffremoved: HashSet<_> = parent_stateinfo
+				let statediffremoved: CompressedState = parent_stateinfo
 					.full_state
 					.difference(&new_state_ids_compressed)
 					.copied()
@@ -409,7 +409,7 @@ impl Service {
 
 				(Arc::new(statediffnew), Arc::new(statediffremoved))
 			} else {
-				(new_state_ids_compressed, Arc::new(HashSet::new()))
+				(new_state_ids_compressed, Arc::new(CompressedState::new()))
 			};
 
 		if !already_existed {
@@ -448,11 +448,11 @@ impl Service {
 			.take_if(|parent| *parent != 0);
 
 		debug_assert!(value.len() % STRIDE == 0, "value not aligned to stride");
-		let num_values = value.len() / STRIDE;
+		let _num_values = value.len() / STRIDE;
 
 		let mut add_mode = true;
-		let mut added = HashSet::with_capacity(num_values);
-		let mut removed = HashSet::with_capacity(num_values);
+		let mut added = CompressedState::new();
+		let mut removed = CompressedState::new();
 
 		let mut i = STRIDE;
 		while let Some(v) = value.get(i..expected!(i + 2 * STRIDE)) {
@@ -469,8 +469,6 @@ impl Service {
 			i = expected!(i + 2 * STRIDE);
 		}
 
-		added.shrink_to_fit();
-		removed.shrink_to_fit();
 		Ok(StateDiff {
 			parent,
 			added: Arc::new(added),
@@ -507,7 +505,7 @@ impl Service {
 
 #[inline]
 #[must_use]
-fn compress_state_event(
+pub(crate) fn compress_state_event(
 	shortstatekey: ShortStateKey,
 	shorteventid: ShortEventId,
 ) -> CompressedStateEvent {
@@ -523,7 +521,7 @@ fn compress_state_event(
 
 #[inline]
 #[must_use]
-pub fn parse_compressed_state_event(
+pub(crate) fn parse_compressed_state_event(
 	compressed_event: CompressedStateEvent,
 ) -> (ShortStateKey, ShortEventId) {
 	use utils::u64_from_u8;

src/service/rooms/timeline/mod.rs

@@ -1,6 +1,7 @@
 mod data;
 
 use std::{
+	borrow::Borrow,
 	cmp,
 	collections::{BTreeMap, HashSet},
 	fmt::Write,
@@ -48,7 +49,7 @@ use crate::{
 	account_data, admin, appservice,
 	appservice::NamespaceRegex,
 	globals, pusher, rooms,
-	rooms::{short::ShortRoomId, state_compressor::CompressedStateEvent},
+	rooms::{short::ShortRoomId, state_compressor::CompressedState},
 	sending, server_keys, users, Dep,
 };
 
@@ -260,14 +261,16 @@ impl Service {
 	///
 	/// Returns pdu id
 	#[tracing::instrument(level = "debug", skip_all)]
-	pub async fn append_pdu(
-		&self,
-		pdu: &PduEvent,
+	pub async fn append_pdu<'a, Leafs>(
+		&'a self,
+		pdu: &'a PduEvent,
 		mut pdu_json: CanonicalJsonObject,
-		leaves: Vec<OwnedEventId>,
-		state_lock: &RoomMutexGuard, /* Take mutex guard to make sure users get the room state
-		                              * mutex */
-	) -> Result<RawPduId> {
+		leafs: Leafs,
+		state_lock: &'a RoomMutexGuard,
+	) -> Result<RawPduId>
+	where
+		Leafs: Iterator<Item = &'a EventId> + Send + 'a,
+	{
 		// Coalesce database writes for the remainder of this scope.
 		let _cork = self.db.db.cork_and_flush();
 
@@ -335,7 +338,7 @@ impl Service {
 
 		self.services
 			.state
-			.set_forward_extremities(&pdu.room_id, leaves, state_lock)
+			.set_forward_extremities(&pdu.room_id, leafs, state_lock)
 			.await;
 
 		let insert_lock = self.mutex_insert.lock(&pdu.room_id).await;
@@ -819,8 +822,7 @@ impl Service {
 		pdu_builder: PduBuilder,
 		sender: &UserId,
 		room_id: &RoomId,
-		state_lock: &RoomMutexGuard, /* Take mutex guard to make sure users get the room state
-		                              * mutex */
+		state_lock: &RoomMutexGuard,
 	) -> Result<OwnedEventId> {
 		let (pdu, pdu_json) = self
 			.create_hash_and_sign_event(pdu_builder, sender, room_id, state_lock)
@@ -896,7 +898,7 @@ impl Service {
 				pdu_json,
 				// Since this PDU references all pdu_leaves we can update the leaves
 				// of the room
-				vec![(*pdu.event_id).to_owned()],
+				once(pdu.event_id.borrow()),
 				state_lock,
 			)
 			.boxed()
@@ -943,16 +945,18 @@ impl Service {
 	/// Append the incoming event setting the state snapshot to the state from
 	/// the server that sent the event.
 	#[tracing::instrument(level = "debug", skip_all)]
-	pub async fn append_incoming_pdu(
-		&self,
-		pdu: &PduEvent,
+	pub async fn append_incoming_pdu<'a, Leafs>(
+		&'a self,
+		pdu: &'a PduEvent,
 		pdu_json: CanonicalJsonObject,
-		new_room_leaves: Vec<OwnedEventId>,
-		state_ids_compressed: Arc<HashSet<CompressedStateEvent>>,
+		new_room_leafs: Leafs,
+		state_ids_compressed: Arc<CompressedState>,
 		soft_fail: bool,
-		state_lock: &RoomMutexGuard, /* Take mutex guard to make sure users get the room state
-		                              * mutex */
-	) -> Result<Option<RawPduId>> {
+		state_lock: &'a RoomMutexGuard,
+	) -> Result<Option<RawPduId>>
+	where
+		Leafs: Iterator<Item = &'a EventId> + Send + 'a,
+	{
 		// We append to state before appending the pdu, so we don't have a moment in
 		// time with the pdu without it's state. This is okay because append_pdu can't
 		// fail.
@@ -968,14 +972,14 @@ impl Service {
 
 			self.services
 				.state
-				.set_forward_extremities(&pdu.room_id, new_room_leaves, state_lock)
+				.set_forward_extremities(&pdu.room_id, new_room_leafs, state_lock)
 				.await;
 
 			return Ok(None);
 		}
 
 		let pdu_id = self
-			.append_pdu(pdu, pdu_json, new_room_leaves, state_lock)
+			.append_pdu(pdu, pdu_json, new_room_leafs, state_lock)
 			.await?;
 
 		Ok(Some(pdu_id))

src/service/sending/sender.rs

@@ -13,7 +13,12 @@ use conduwuit::{
 	debug, err, error,
 	result::LogErr,
 	trace,
-	utils::{calculate_hash, continue_exponential_backoff_secs, stream::IterStream, ReadyExt},
+	utils::{
+		calculate_hash, continue_exponential_backoff_secs,
+		future::TryExtExt,
+		stream::{BroadbandExt, IterStream, WidebandExt},
+		ReadyExt,
+	},
 	warn, Error, Result,
 };
 use futures::{
@@ -62,8 +67,6 @@ type SendingFuture<'a> = BoxFuture<'a, SendingResult>;
 type SendingFutures<'a> = FuturesUnordered<SendingFuture<'a>>;
 type CurTransactionStatus = HashMap<Destination, TransactionStatus>;
 
-const CLEANUP_TIMEOUT_MS: u64 = 3500;
-
 const SELECT_PRESENCE_LIMIT: usize = 256;
 const SELECT_RECEIPT_LIMIT: usize = 256;
 const SELECT_EDU_LIMIT: usize = EDU_LIMIT - 2;
@@ -211,8 +214,9 @@ impl Service {
 			time::{sleep_until, Instant},
 		};
 
+		let timeout = self.server.config.sender_shutdown_timeout;
+		let timeout = Duration::from_secs(timeout);
 		let now = Instant::now();
-		let timeout = Duration::from_millis(CLEANUP_TIMEOUT_MS);
 		let deadline = now.checked_add(timeout).unwrap_or(now);
 		loop {
 			trace!("Waiting for {} requests to complete...", futures.len());
@@ -474,20 +478,25 @@ impl Service {
 		since: (u64, u64),
 		max_edu_count: &AtomicU64,
 	) -> Option<EduBuf> {
-		let server_rooms = self.services.state_cache.server_rooms(server_name);
-
-		pin_mut!(server_rooms);
 		let mut num = 0;
-		let mut receipts = BTreeMap::<OwnedRoomId, ReceiptMap>::new();
-		while let Some(room_id) = server_rooms.next().await {
-			let receipt_map = self
-				.select_edus_receipts_room(room_id, since, max_edu_count, &mut num)
-				.await;
+		let receipts: BTreeMap<OwnedRoomId, ReceiptMap> = self
+			.services
+			.state_cache
+			.server_rooms(server_name)
+			.map(ToOwned::to_owned)
+			.broad_filter_map(|room_id| async move {
+				let receipt_map = self
+					.select_edus_receipts_room(&room_id, since, max_edu_count, &mut num)
+					.await;
 
-			if !receipt_map.read.is_empty() {
-				receipts.insert(room_id.into(), receipt_map);
-			}
-		}
+				receipt_map
+					.read
+					.is_empty()
+					.eq(&false)
+					.then_some((room_id, receipt_map))
+			})
+			.collect()
+			.await;
 
 		if receipts.is_empty() {
 			return None;
@@ -820,9 +829,8 @@ impl Service {
 				| _ => None,
 			})
 			.stream()
-			.then(|pdu_id| self.services.timeline.get_pdu_json_from_id(pdu_id))
-			.ready_filter_map(Result::ok)
-			.then(|pdu| self.convert_to_outgoing_federation_event(pdu))
+			.wide_filter_map(|pdu_id| self.services.timeline.get_pdu_json_from_id(pdu_id).ok())
+			.wide_then(|pdu| self.convert_to_outgoing_federation_event(pdu))
 			.collect()
 			.await;
 

src/service/uiaa/mod.rs

@@ -1,5 +1,5 @@
 use std::{
-	collections::BTreeMap,
+	collections::{BTreeMap, HashSet},
 	sync::{Arc, RwLock},
 };
 
@@ -17,7 +17,7 @@ use ruma::{
 	CanonicalJsonValue, DeviceId, OwnedDeviceId, OwnedUserId, UserId,
 };
 
-use crate::{globals, users, Dep};
+use crate::{config, globals, users, Dep};
 
 pub struct Service {
 	userdevicesessionid_uiaarequest: RwLock<RequestMap>,
@@ -28,6 +28,7 @@ pub struct Service {
 struct Services {
 	globals: Dep<globals::Service>,
 	users: Dep<users::Service>,
+	config: Dep<config::Service>,
 }
 
 struct Data {
@@ -49,6 +50,7 @@ impl crate::Service for Service {
 			services: Services {
 				globals: args.depend::<globals::Service>("globals"),
 				users: args.depend::<users::Service>("users"),
+				config: args.depend::<config::Service>("config"),
 			},
 		}))
 	}
@@ -56,6 +58,26 @@ impl crate::Service for Service {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
+#[implement(Service)]
+pub async fn read_tokens(&self) -> Result<HashSet<String>> {
+	let mut tokens = HashSet::new();
+	if let Some(file) = &self.services.config.registration_token_file.as_ref() {
+		match std::fs::read_to_string(file) {
+			| Ok(text) => {
+				text.split_ascii_whitespace().for_each(|token| {
+					tokens.insert(token.to_owned());
+				});
+			},
+			| Err(e) => error!("Failed to read the registration token file: {e}"),
+		}
+	};
+	if let Some(token) = &self.services.config.registration_token {
+		tokens.insert(token.to_owned());
+	}
+
+	Ok(tokens)
+}
+
 /// Creates a new Uiaa session. Make sure the session token is unique.
 #[implement(Service)]
 pub fn create(
@@ -152,13 +174,8 @@ pub async fn try_auth(
 			uiaainfo.completed.push(AuthType::Password);
 		},
 		| AuthData::RegistrationToken(t) => {
-			if self
-				.services
-				.globals
-				.registration_token
-				.as_ref()
-				.is_some_and(|reg_token| t.token.trim() == reg_token)
-			{
+			let tokens = self.read_tokens().await?;
+			if tokens.contains(t.token.trim()) {
 				uiaainfo.completed.push(AuthType::RegistrationToken);
 			} else {
 				uiaainfo.auth_error = Some(ruma::api::client::error::StandardErrorBody {

src/service/users/mod.rs

@@ -1,12 +1,12 @@
-use std::{collections::BTreeMap, mem, mem::size_of, sync::Arc};
+use std::{collections::BTreeMap, mem, sync::Arc};
 
 use conduwuit::{
-	debug_warn, err, trace,
+	at, debug_warn, err, trace,
 	utils::{self, stream::TryIgnore, string::Unquoted, ReadyExt},
 	Err, Error, Result, Server,
 };
-use database::{Database, Deserialized, Ignore, Interfix, Json, Map};
-use futures::{FutureExt, Stream, StreamExt, TryFutureExt};
+use database::{Deserialized, Ignore, Interfix, Json, Map};
+use futures::{Stream, StreamExt, TryFutureExt};
 use ruma::{
 	api::client::{device::Device, error::ErrorKind, filter::FilterDefinition},
 	encryption::{CrossSigningKey, DeviceKeys, OneTimeKey},
@@ -28,7 +28,6 @@ pub struct Service {
 
 struct Services {
 	server: Arc<Server>,
-	db: Arc<Database>,
 	account_data: Dep<account_data::Service>,
 	admin: Dep<admin::Service>,
 	globals: Dep<globals::Service>,
@@ -64,7 +63,6 @@ impl crate::Service for Service {
 		Ok(Arc::new(Self {
 			services: Services {
 				server: args.server.clone(),
-				db: args.db.clone(),
 				account_data: args.depend::<account_data::Service>("account_data"),
 				admin: args.depend::<admin::Service>("admin"),
 				globals: args.depend::<globals::Service>("globals"),
@@ -792,44 +790,47 @@ impl Service {
 		&'a self,
 		user_id: &'a UserId,
 		device_id: &'a DeviceId,
+		since: Option<u64>,
+		to: Option<u64>,
 	) -> impl Stream<Item = Raw<AnyToDeviceEvent>> + Send + 'a {
-		let prefix = (user_id, device_id, Interfix);
+		type Key<'a> = (&'a UserId, &'a DeviceId, u64);
+
+		let from = (user_id, device_id, since.map_or(0, |since| since.saturating_add(1)));
+
 		self.db
 			.todeviceid_events
-			.stream_prefix(&prefix)
+			.stream_from(&from)
 			.ignore_err()
-			.map(|(_, val): (Ignore, Raw<AnyToDeviceEvent>)| val)
+			.ready_take_while(move |((user_id_, device_id_, count), _): &(Key<'_>, _)| {
+				user_id == *user_id_
+					&& device_id == *device_id_
+					&& to.is_none_or(|to| *count <= to)
+			})
+			.map(at!(1))
 	}
 
-	pub async fn remove_to_device_events(
+	pub async fn remove_to_device_events<Until>(
 		&self,
 		user_id: &UserId,
 		device_id: &DeviceId,
-		until: u64,
-	) {
-		let mut prefix = user_id.as_bytes().to_vec();
-		prefix.push(0xFF);
-		prefix.extend_from_slice(device_id.as_bytes());
-		prefix.push(0xFF);
+		until: Until,
+	) where
+		Until: Into<Option<u64>> + Send,
+	{
+		type Key<'a> = (&'a UserId, &'a DeviceId, u64);
 
-		let mut last = prefix.clone();
-		last.extend_from_slice(&until.to_be_bytes());
-
-		let _cork = self.services.db.cork_and_flush();
+		let until = until.into().unwrap_or(u64::MAX);
+		let from = (user_id, device_id, until);
 		self.db
 			.todeviceid_events
-			.rev_raw_keys_from(&last) // this includes last
+			.rev_keys_from(&from)
 			.ignore_err()
-			.ready_take_while(move |key| key.starts_with(&prefix))
-			.map(|key| {
-				let len = key.len();
-				let start = len.saturating_sub(size_of::<u64>());
-				let count = utils::u64_from_u8(&key[start..len]);
-				(key, count)
+			.ready_take_while(move |(user_id_, device_id_, _): &Key<'_>| {
+				user_id == *user_id_ && device_id == *device_id_
+			})
+			.ready_for_each(|key: Key<'_>| {
+				self.db.todeviceid_events.del(key);
 			})
-			.ready_take_while(move |(_, count)| *count <= until)
-			.ready_for_each(|(key, _)| self.db.todeviceid_events.remove(&key))
-			.boxed()
 			.await;
 	}