fix: Don't bail out when changing profile data if membership cache is wrong

works around build issue:
https://github.com/aws/aws-lc-rs/issues/1168

Cargo.lock

@@ -97,15 +97,15 @@ checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000"
 
 [[package]]
 name = "anyhow"
-version = "1.0.102"
+version = "1.0.103"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
+checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3"
 
 [[package]]
 name = "arc-swap"
-version = "1.9.1"
+version = "1.9.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207"
+checksum = "c049c0be4daef0b145cb3555416b3b8ef5b7888a38aea1a3a155801fe7b0810b"
 dependencies = [
  "rustversion",
 ]
@@ -259,9 +259,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.17.1"
+version = "1.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4342d8937fc7e5dd9b1c60292261c0670c882a2cd1719cfc11b1af41731e32ad"
+checksum = "5ec2f1fc3ec205783a5da9a7e6c1509cc69dedf09a1949e412c1e18469326d00"
 dependencies = [
  "aws-lc-sys",
  "zeroize",
@@ -269,15 +269,14 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.42.0"
+version = "0.41.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d9ceb1da931507a12f4fccea479dccd00da1943e1b4ae72d8e502d707361444"
+checksum = "1a2f9779ce85b93ab6170dd940ad0169b5766ff848247aff13bb788b832fe3f4"
 dependencies = [
  "cc",
  "cmake",
  "dunce",
  "fs_extra",
- "pkg-config",
 ]
 
 [[package]]
@@ -598,9 +597,9 @@ dependencies = [
 
 [[package]]
 name = "camino"
-version = "1.2.2"
+version = "1.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e629a66d692cb9ff1a1c664e41771b3dcaf961985a9774c0eb0bd1b51cf60a48"
+checksum = "5f2d30e4173c4026932d51d31d6b0613b1fd3014bf3f9f8943d4ba139c437ba0"
 dependencies = [
  "serde_core",
 ]
@@ -642,9 +641,9 @@ dependencies = [
 
 [[package]]
 name = "cc"
-version = "1.2.64"
+version = "1.2.65"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dad887fd958be91b5098c0248def011f4523ab786cd411be668777e55063501f"
+checksum = "e228eec9be7c17ccb640b59b36a5cd805ea2a564a4c5e162c2f659fea30d3b96"
 dependencies = [
  "find-msvc-tools",
  "jobserver",
@@ -675,9 +674,9 @@ checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
 
 [[package]]
 name = "chacha20"
-version = "0.10.0"
+version = "0.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601"
+checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81"
 dependencies = [
  "cfg-if",
  "cpufeatures 0.3.0",
@@ -880,6 +879,7 @@ dependencies = [
  "futures",
  "lettre",
  "log",
+ "resolvematrix",
  "ruma",
  "serde-saphyr",
  "serde_json",
@@ -979,6 +979,7 @@ dependencies = [
  "rand_core 0.6.4",
  "regex",
  "reqwest 0.13.4",
+ "resolvematrix",
  "ruma",
  "sanitize-filename",
  "serde",
@@ -1105,6 +1106,7 @@ dependencies = [
  "regex",
  "reqwest 0.12.28",
  "reqwest 0.13.4",
+ "resolvematrix",
  "ruma",
  "ruminuwuity",
  "rustyline-async",
@@ -1442,9 +1444,9 @@ dependencies = [
 
 [[package]]
 name = "crypto-common"
-version = "0.1.7"
+version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
+checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
 dependencies = [
  "generic-array",
  "typenum",
@@ -1652,7 +1654,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
  "block-buffer 0.10.4",
  "const-oid",
- "crypto-common 0.1.7",
+ "crypto-common 0.1.6",
  "subtle",
 ]
 
@@ -1940,12 +1942,6 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
-[[package]]
-name = "foldhash"
-version = "0.1.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
-
 [[package]]
 name = "foldhash"
 version = "0.2.0"
@@ -1963,9 +1959,9 @@ dependencies = [
 
 [[package]]
 name = "fs-err"
-version = "3.3.0"
+version = "3.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "73fde052dbfc920003cfd2c8e2c6e6d4cc7c1091538c3a24226cec0665ab08c0"
+checksum = "b91aa448ca50d7e79433bdf3ee8d99215430d2ec02ade5aefab2a073a1822e8a"
 dependencies = [
  "autocfg",
  "tokio",
@@ -2083,9 +2079,9 @@ dependencies = [
 
 [[package]]
 name = "generic-array"
-version = "0.14.7"
+version = "0.14.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
+checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2"
 dependencies = [
  "typenum",
  "version_check",
@@ -2121,16 +2117,14 @@ dependencies = [
 
 [[package]]
 name = "getrandom"
-version = "0.4.2"
+version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555"
+checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099"
 dependencies = [
  "cfg-if",
  "libc",
  "r-efi 6.0.0",
  "rand_core 0.10.1",
- "wasip2",
- "wasip3",
 ]
 
 [[package]]
@@ -2254,15 +2248,6 @@ version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"
 
-[[package]]
-name = "hashbrown"
-version = "0.15.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
-dependencies = [
- "foldhash 0.1.5",
-]
-
 [[package]]
 name = "hashbrown"
 version = "0.16.1"
@@ -2271,7 +2256,7 @@ checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
 dependencies = [
  "allocator-api2",
  "equivalent",
- "foldhash 0.2.0",
+ "foldhash",
 ]
 
 [[package]]
@@ -2515,15 +2500,15 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
 
 [[package]]
 name = "humantime"
-version = "2.3.0"
+version = "2.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424"
+checksum = "15cdd26707701c53297e2fa6afb323d55fbc1d0810c3aec078ae3ef0424c3c15"
 
 [[package]]
 name = "hybrid-array"
-version = "0.4.12"
+version = "0.4.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9155a582abd142abc056962c29e3ce5ff2ad5469f4246b537ed42c5deba857da"
+checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c"
 dependencies = [
  "typenum",
 ]
@@ -2708,12 +2693,6 @@ dependencies = [
  "zerovec",
 ]
 
-[[package]]
-name = "id-arena"
-version = "2.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954"
-
 [[package]]
 name = "ident_case"
 version = "1.0.1"
@@ -2937,9 +2916,9 @@ dependencies = [
 
 [[package]]
 name = "js-sys"
-version = "0.3.102"
+version = "0.3.103"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "03d04c30968dffe80775bd4d7fb676131cd04a1fb46d2686dbffbaec2d9dfd31"
+checksum = "53b44bfcdb3f8d5837a46dae1ca9660a837176eee74a28b229bc626816589102"
 dependencies = [
  "cfg-if",
  "futures-util",
@@ -3006,12 +2985,6 @@ dependencies = [
  "spin",
 ]
 
-[[package]]
-name = "leb128fmt"
-version = "0.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2"
-
 [[package]]
 name = "lettre"
 version = "0.11.22"
@@ -3087,9 +3060,9 @@ dependencies = [
 
 [[package]]
 name = "link-section"
-version = "0.18.2"
+version = "0.18.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c2b1dd6fe32e55c0fc0ea9493aa57459ca3cf4ff3c857c7d0302290150da6e4f"
+checksum = "24670b639492630905459a6c7d47f063d33c2d4fcd5362f6e5827c5613976c9f"
 
 [[package]]
 name = "linked-hash-map"
@@ -3422,9 +3395,9 @@ dependencies = [
 
 [[package]]
 name = "num-bigint"
-version = "0.4.6"
+version = "0.4.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9"
+checksum = "c863e9ab5e7bf9c99ba75e1050f1e4d624ae87ed3532d6238ffbdc7b585dbbe6"
 dependencies = [
  "num-integer",
  "num-traits",
@@ -4120,16 +4093,6 @@ dependencies = [
  "num-traits",
 ]
 
-[[package]]
-name = "prettyplease"
-version = "0.2.37"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b"
-dependencies = [
- "proc-macro2",
- "syn",
-]
-
 [[package]]
 name = "primeorder"
 version = "0.13.6"
@@ -4256,9 +4219,9 @@ checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3"
 
 [[package]]
 name = "quinn"
-version = "0.11.9"
+version = "0.11.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9e20a958963c291dc322d98411f541009df2ced7b5a4f2bd52337638cfccf20"
+checksum = "0c1a41e437b6bbd489372cd4971de128e85c855f56c57f283d20ff016cf7c0a8"
 dependencies = [
  "bytes",
  "cfg_aliases",
@@ -4277,9 +4240,9 @@ dependencies = [
 
 [[package]]
 name = "quinn-proto"
-version = "0.11.14"
+version = "0.11.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098"
+checksum = "4fcb935c5bec503c2f0e306bdd3e58bb9029dcb14fa8d9ac76e3a5256ac0763e"
 dependencies = [
  "aws-lc-rs",
  "bytes",
@@ -4366,7 +4329,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80"
 dependencies = [
  "chacha20",
- "getrandom 0.4.2",
+ "getrandom 0.4.3",
  "rand_core 0.10.1",
 ]
 
@@ -4574,6 +4537,23 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7"
 
+[[package]]
+name = "resolvematrix"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "318f1067d605ee34de83fabc0f0b73e8664b580abea8d5e45742fab534e2b18a"
+dependencies = [
+ "futures",
+ "hickory-resolver",
+ "num-traits",
+ "parking_lot",
+ "reqwest 0.13.4",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.18",
+ "tracing",
+]
+
 [[package]]
 name = "rfc6979"
 version = "0.4.0"
@@ -4681,7 +4661,7 @@ dependencies = [
  "bytes",
  "date_header",
  "form_urlencoded",
- "getrandom 0.4.2",
+ "getrandom 0.4.3",
  "http",
  "indexmap 2.14.0",
  "js_int",
@@ -4862,9 +4842,9 @@ checksum = "b50b8869d9fc858ce7266cce0194bd74df58b9d0e3f6df3a9fc8eb470d95c09d"
 
 [[package]]
 name = "rustc-hash"
-version = "2.1.2"
+version = "2.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe"
+checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d"
 
 [[package]]
 name = "rustc_version"
@@ -4918,9 +4898,9 @@ dependencies = [
 
 [[package]]
 name = "rustls-pki-types"
-version = "1.14.1"
+version = "1.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9"
+checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046"
 dependencies = [
  "web-time",
  "zeroize",
@@ -5844,9 +5824,9 @@ dependencies = [
 
 [[package]]
 name = "time"
-version = "0.3.49"
+version = "0.3.53"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "711a53c2d47bbd818258c498c8dbfe186a2526c631495cfe7e078567f86b8469"
+checksum = "18dfaaeddcb932337b5e7866ee7d0ce9b76d2fd092997146f187ec09b4558a50"
 dependencies = [
  "deranged",
  "num-conv",
@@ -5864,9 +5844,9 @@ checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109"
 
 [[package]]
 name = "time-macros"
-version = "0.2.29"
+version = "0.2.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "71c652a3727a9cbb9a02f707f530b618ce00d0ccd762009c8c23bd191df3c17d"
+checksum = "c431b87111666e491a90baa837f914fb45cd5dc3c268591b0220ff5057f2085f"
 dependencies = [
  "num-conv",
  "time-core",
@@ -6424,12 +6404,6 @@ version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b4ac048d71ede7ee76d585517add45da530660ef4390e49b098733c6e897f254"
 
-[[package]]
-name = "unicode-xid"
-version = "0.2.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
-
 [[package]]
 name = "untrusted"
 version = "0.9.0"
@@ -6469,11 +6443,11 @@ checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be"
 
 [[package]]
 name = "uuid"
-version = "1.23.3"
+version = "1.23.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "144d6b123cef80b301b8f72a9e2ca4370ddec21950d0a103dd22c437006d2db7"
+checksum = "bf80a72845275afea99e7f2b434723d3bc7e38470fcd1c7ed39a599c73319a53"
 dependencies = [
- "getrandom 0.4.2",
+ "getrandom 0.4.3",
  "js-sys",
  "serde_core",
  "wasm-bindgen",
@@ -6558,23 +6532,14 @@ version = "1.0.4+wasi-0.2.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487"
 dependencies = [
- "wit-bindgen 0.57.1",
-]
-
-[[package]]
-name = "wasip3"
-version = "0.4.0+wasi-0.3.0-rc-2026-01-06"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5"
-dependencies = [
- "wit-bindgen 0.51.0",
+ "wit-bindgen",
 ]
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.125"
+version = "0.2.126"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8ddb3f79143bced6de84270411622a2699cee572fc0875aeaf1e7867cf9fca1a"
+checksum = "4b067c0c11094aef6b7a801c1e34a26affafdf3d051dba08456b868789aaf9a4"
 dependencies = [
  "cfg-if",
  "once_cell",
@@ -6585,9 +6550,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-futures"
-version = "0.4.75"
+version = "0.4.76"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "503b14d284f2c8dac03b819967e155ea753f573586193b2b2c95990cb5d69280"
+checksum = "c62df1340f32221cb9c54d6a27b030e3dba64361d4a95bed55f9aacb44da291d"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -6595,9 +6560,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.125"
+version = "0.2.126"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4e21a184b13fb19e157296e2c46056aec9092264fab83e4ba59e68c61b323c3d"
+checksum = "167ce5e579f6bcf889c4f7175a8a5a585de84e8ff93976ce393efa5f2837aab1"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -6605,9 +6570,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.125"
+version = "0.2.126"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fecefd9c35bd935a20fc3fc344b5f29138961e4f47fb03297d88f2587afb5ebd"
+checksum = "f3997c7839262f4ef12cf90b818d6340c18e80f263f1a94bf157d0ec4420380e"
 dependencies = [
  "bumpalo",
  "proc-macro2",
@@ -6618,35 +6583,13 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.125"
+version = "0.2.126"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "23939e44bb9a5d7576fa2b563dc2e136628f1224e88a8deed09e04858b77871f"
+checksum = "dc1b4cb0cc549fcf58d7dfc081778139b3d283a081644e833e84682ad71cea24"
 dependencies = [
  "unicode-ident",
 ]
 
-[[package]]
-name = "wasm-encoder"
-version = "0.244.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "990065f2fe63003fe337b932cfb5e3b80e0b4d0f5ff650e6985b1048f62c8319"
-dependencies = [
- "leb128fmt",
- "wasmparser",
-]
-
-[[package]]
-name = "wasm-metadata"
-version = "0.244.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909"
-dependencies = [
- "anyhow",
- "indexmap 2.14.0",
- "wasm-encoder",
- "wasmparser",
-]
-
 [[package]]
 name = "wasm-streams"
 version = "0.5.0"
@@ -6660,23 +6603,11 @@ dependencies = [
  "web-sys",
 ]
 
-[[package]]
-name = "wasmparser"
-version = "0.244.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe"
-dependencies = [
- "bitflags",
- "hashbrown 0.15.5",
- "indexmap 2.14.0",
- "semver",
-]
-
 [[package]]
 name = "web-sys"
-version = "0.3.102"
+version = "0.3.103"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a6430a72df5eb332242960fe84b3002a241163998241eb596d4f739b9757061d"
+checksum = "8622dcb61c0bcc9fffa6938bed81210af2da9a7e4a1a834b2e37a59b6dfb6141"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -6706,9 +6637,9 @@ dependencies = [
 
 [[package]]
 name = "webpki-root-certs"
-version = "1.0.7"
+version = "1.0.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f31141ce3fc3e300ae89b78c0dd67f9708061d1d2eda54b8209346fd6be9a92c"
+checksum = "0d46a5a140e6f7afeccd8eae97eff335163939eac8b929834875168b29b3d267"
 dependencies = [
  "rustls-pki-types",
 ]
@@ -7015,100 +6946,12 @@ dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "wit-bindgen"
-version = "0.51.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5"
-dependencies = [
- "wit-bindgen-rust-macro",
-]
-
 [[package]]
 name = "wit-bindgen"
 version = "0.57.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
 
-[[package]]
-name = "wit-bindgen-core"
-version = "0.51.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ea61de684c3ea68cb082b7a88508a8b27fcc8b797d738bfc99a82facf1d752dc"
-dependencies = [
- "anyhow",
- "heck",
- "wit-parser",
-]
-
-[[package]]
-name = "wit-bindgen-rust"
-version = "0.51.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21"
-dependencies = [
- "anyhow",
- "heck",
- "indexmap 2.14.0",
- "prettyplease",
- "syn",
- "wasm-metadata",
- "wit-bindgen-core",
- "wit-component",
-]
-
-[[package]]
-name = "wit-bindgen-rust-macro"
-version = "0.51.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c0f9bfd77e6a48eccf51359e3ae77140a7f50b1e2ebfe62422d8afdaffab17a"
-dependencies = [
- "anyhow",
- "prettyplease",
- "proc-macro2",
- "quote",
- "syn",
- "wit-bindgen-core",
- "wit-bindgen-rust",
-]
-
-[[package]]
-name = "wit-component"
-version = "0.244.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2"
-dependencies = [
- "anyhow",
- "bitflags",
- "indexmap 2.14.0",
- "log",
- "serde",
- "serde_derive",
- "serde_json",
- "wasm-encoder",
- "wasm-metadata",
- "wasmparser",
- "wit-parser",
-]
-
-[[package]]
-name = "wit-parser"
-version = "0.244.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736"
-dependencies = [
- "anyhow",
- "id-arena",
- "indexmap 2.14.0",
- "log",
- "semver",
- "serde",
- "serde_derive",
- "serde_json",
- "unicode-xid",
- "wasmparser",
-]
-
 [[package]]
 name = "writeable"
 version = "0.6.3"

Cargo.toml

@@ -141,6 +141,12 @@ features = [
 version = "0.23.25"
 default-features = false
 
+[workspace.dependencies.aws-lc-sys]
+version = "0.41.0"
+
+[workspace.dependencies.aws-lc-rs]
+version = "1.17.0"
+
 [workspace.dependencies.reqwest]
 version = "0.13.2"
 default-features = false
@@ -563,6 +569,9 @@ features = ["std"]
 [workspace.dependencies.nonzero_ext]
 version = "0.3.0"
 
+[workspace.dependencies.resolvematrix]
+version = "0.1.0"
+
 [workspace.dependencies.serde_urlencoded]
 version = "0.7.1"
 

changelog.d/+32bb7e0d.bugfix.md

@@ -0,0 +1 @@
+Improved invite and join reliability in clients using legacy sync. Contributed by @ginger

changelog.d/1505.refactor.md

@@ -0,0 +1 @@
+Rewrite the resolver service to use [resolvematrix](https://forgejo.ellis.link/continuwuation/resolvematrix) for server resolution. Rewrite by @s1lv3r, crate by @Jade

src/admin/Cargo.toml

@@ -92,6 +92,7 @@ serde-saphyr.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true
+resolvematrix.workspace = true
 
 [lints]
 workspace = true

src/admin/debug/commands.rs

@@ -20,6 +20,10 @@
 };
 use futures::{FutureExt, StreamExt, TryStreamExt};
 use lettre::message::Mailbox;
+use resolvematrix::{
+	resolution::ResolvedDestination,
+	server::{MatrixResolver, MatrixResolverBuilder},
+};
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId,
 	OwnedRoomOrAliasId, OwnedServerName, RoomId, RoomVersionId, UInt,
@@ -1007,13 +1011,23 @@ pub(super) async fn resolve_true_destination(
 			);
 		}
 
-		let actual = self
-			.services
-			.resolver
-			.resolve_actual_dest(&server_name, !no_cache)
-			.await?;
+		let resolver: &MatrixResolver = if no_cache {
+			&MatrixResolverBuilder::new()
+				.dangerous_tls_accept_invalid_certs(self.services.server.config.allow_invalid_tls_certificates_yes_i_know_what_the_fuck_i_am_doing_with_this_and_i_know_this_is_insecure)
+				.http_client(self.services.client.default.clone())
+				.build()?
+		} else {
+			&self.services.resolver.resolver
+		};
 
-		let msg = format!("Destination: {}\nHostname URI: {}", actual.dest, actual.host);
+		let actual = resolver.resolve_server(server_name.as_str()).await?;
+
+		let destination = match actual.destination {
+			| ResolvedDestination::Literal(addr) => addr.to_string(),
+			| ResolvedDestination::Named(host, port) => format!("{host}:{port}"),
+		};
+
+		let msg = format!("Destination: {}\nHostname URI (SNI): {}", destination, actual.host);
 		self.write_str(&msg).await
 	}
 

src/admin/query/resolver.rs

@@ -46,7 +46,7 @@ async fn destinations_cache(&self, server_name: Option<OwnedServerName>) -> Resu
 		writeln!(self, "| Server Name | Destination | Hostname | Expires |").await?;
 		writeln!(self, "| ----------- | ----------- | -------- | ------- |").await?;
 
-		let mut destinations = self.services.resolver.cache.destinations().boxed();
+		let mut destinations = self.services.resolver.dns.cache.destinations().boxed();
 
 		while let Some((name, CachedDest { dest, host, expire })) = destinations.next().await {
 			if let Some(server_name) = server_name.as_ref() {
@@ -69,7 +69,7 @@ async fn overrides_cache(&self, server_name: Option<String>) -> Result {
 		writeln!(self, "| Server Name | IP  | Port | Expires | Overriding |").await?;
 		writeln!(self, "| ----------- | --- | ----:| ------- | ---------- |").await?;
 
-		let mut overrides = self.services.resolver.cache.overrides().boxed();
+		let mut overrides = self.services.resolver.dns.cache.overrides().boxed();
 
 		while let Some((name, CachedOverride { ips, port, expire, overriding })) =
 			overrides.next().await
@@ -92,11 +92,16 @@ async fn overrides_cache(&self, server_name: Option<String>) -> Result {
 
 	async fn flush_cache(&self, name: Option<OwnedServerName>, all: bool) -> Result {
 		if all {
-			self.services.resolver.cache.clear().await;
+			self.services.resolver.resolver.clear_cache();
+			self.services.resolver.dns.cache.clear().await;
 			writeln!(self, "Resolver caches cleared!").await
 		} else if let Some(name) = name {
-			self.services.resolver.cache.del_destination(&name);
-			self.services.resolver.cache.del_override(&name);
+			self.services
+				.resolver
+				.resolver
+				.remove_cache_entry(name.as_str());
+			self.services.resolver.dns.cache.del_destination(&name);
+			self.services.resolver.dns.cache.del_override(&name);
 			self.write_str(&format!("Cleared {name} from resolver caches!"))
 				.await
 		} else {

src/api/client/membership/leave.rs

@@ -170,9 +170,7 @@ pub async fn leave_room(
 					 locally."
 				);
 
-				// return the existing leave state, if one exists. `mark_as_left` will then
-				// update the `roomuserid_leftcount` table, making the leave come down sync
-				// again.
+				// return the existing leave state, if one exists
 				services
 					.rooms
 					.state_cache
@@ -207,6 +205,8 @@ pub async fn leave_room(
 		.update_joined_count(room_id)
 		.await;
 
+	services.sync.wake(user_id).await;
+
 	Ok(())
 }
 

src/api/client/read_marker.rs

@@ -107,6 +107,8 @@ pub(crate) async fn set_read_marker_route(
 			.private_read_set(&body.room_id, sender_user, count);
 	}
 
+	services.sync.wake(sender_user).await;
+
 	Ok(set_read_marker::v3::Response::new())
 }
 
@@ -209,5 +211,7 @@ pub(crate) async fn create_receipt_route(
 		},
 	}
 
+	services.sync.wake(sender_user).await;
+
 	Ok(create_receipt::v3::Response::new())
 }

src/api/client/sync/v3/mod.rs

@@ -201,9 +201,6 @@ pub(crate) async fn sync_events_route(
 		.update_device_last_seen(sender_user, Some(sender_device), client_ip)
 		.await;
 
-	// Setup watchers, so if there's no response, we can wait for them
-	let watcher = services.sync.watch(sender_user, sender_device);
-
 	let response = build_sync_events(&services, &body).await?;
 	if body.body.full_state
 		|| !(response.rooms.is_empty()
@@ -219,7 +216,7 @@ pub(crate) async fn sync_events_route(
 	// Stop hanging if new info arrives
 	let default = Duration::from_secs(30);
 	let duration = cmp::min(body.body.timeout.unwrap_or(default), default);
-	_ = tokio::time::timeout(duration, watcher).await;
+	_ = tokio::time::timeout(duration, services.sync.wait_for_wake(sender_user)).await;
 
 	// Retry returning data
 	build_sync_events(&services, &body).await

src/api/client/sync/v5.rs

@@ -80,9 +80,6 @@ pub(crate) async fn sync_events_v5_route(
 
 	let mut body = body.body;
 
-	// Setup watchers, so if there's no response, we can wait for them
-	let watcher = services.sync.watch(sender_user, sender_device);
-
 	let next_batch = services.globals.next_count()?;
 
 	let conn_id = body.conn_id.clone();
@@ -220,7 +217,7 @@ pub(crate) async fn sync_events_v5_route(
 		// Stop hanging if new info arrives
 		let default = Duration::from_secs(30);
 		let duration = cmp::min(body.timeout.unwrap_or(default), default);
-		_ = tokio::time::timeout(duration, watcher).await;
+		_ = tokio::time::timeout(duration, services.sync.wait_for_wake(sender_user)).await;
 	}
 
 	let typing = collect_typing_events(services, sender_user, &body, &todo_rooms).await?;

src/api/router/auth.rs

@@ -60,8 +60,9 @@ pub(crate) fn sender_device(&self) -> Option<&DeviceId> {
 
 	pub(crate) fn expect_sender_device(&self) -> Result<&DeviceId> {
 		match self {
-			| Self::User { sender_device, .. } => Ok(sender_device),
-			| Self::Appservice { .. } =>
+			| Self::User { sender_device, .. }
+			| Self::Appservice { sender_device: Some(sender_device), .. } => Ok(sender_device),
+			| Self::Appservice { sender_device: None, .. } =>
 				Err!(Request(Forbidden("Appservices must masquerade to use this endpoint."))),
 		}
 	}

src/api/server/invite.rs

@@ -203,6 +203,8 @@ pub(crate) async fn create_invite_route(
 			.update_joined_count(&body.room_id)
 			.await;
 
+		services.sync.wake(&recipient_user).await;
+
 		for appservice in services.appservice.read().await.values() {
 			if appservice.is_user_match(&recipient_user) {
 				let transaction_id = general_purpose::URL_SAFE_NO_PAD

src/core/Cargo.toml

@@ -118,6 +118,7 @@ url.workspace = true
 parking_lot.workspace = true
 lock_api.workspace = true
 hyper-util.workspace = true
+resolvematrix.workspace = true
 openidconnect.workspace = true
 
 [target.'cfg(unix)'.dependencies]

src/core/error/mod.rs

@@ -87,6 +87,8 @@ pub enum Error {
 	YamlDe(#[from] serde_saphyr::Error),
 	#[error(transparent)]
 	YamlSer(#[from] serde_saphyr::ser_error::Error),
+	#[error(transparent)]
+	ResolveServer(#[from] resolvematrix::error::ResolveServerError),
 
 	// ruma/conduwuit
 	#[error("Arithmetic operation failed: {0}")]

src/service/Cargo.toml

@@ -119,6 +119,7 @@ recaptcha-verify = { version = "0.2.0", default-features = false }
 reqwest_recaptcha = { package = "reqwest", version = "0.12.28", default-features = false, features = ["rustls-tls-native-roots-no-provider"]  } # As long as recaptcha-verify's reqwest is outdated
 yansi.workspace = true
 lettre.workspace = true
+resolvematrix.workspace = true
 serde_urlencoded.workspace = true
 openidconnect.workspace = true
 

src/service/account_data/mod.rs

@@ -16,7 +16,7 @@
 };
 use serde::Deserialize;
 
-use crate::{Dep, globals};
+use crate::{Dep, globals, sync};
 
 #[derive(Debug)]
 pub enum AnyRawAccountDataEvent {
@@ -36,6 +36,7 @@ struct Data {
 
 struct Services {
 	globals: Dep<globals::Service>,
+	sync: Dep<sync::Service>,
 }
 
 impl crate::Service for Service {
@@ -43,6 +44,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
 			services: Services {
 				globals: args.depend::<globals::Service>("globals"),
+				sync: args.depend::<sync::Service>("sync"),
 			},
 			db: Data {
 				roomuserdataid_accountdata: args.db["roomuserdataid_accountdata"].clone(),
@@ -84,6 +86,8 @@ pub async fn update(
 			self.db.roomuserdataid_accountdata.remove(&prev);
 		}
 
+		self.services.sync.wake(user_id).await;
+
 		Ok(())
 	}
 

src/service/client/mod.rs

@@ -11,7 +11,6 @@ pub struct Service {
 	pub default: reqwest::Client,
 	pub url_preview: reqwest::Client,
 	pub extern_media: reqwest::Client,
-	pub well_known: reqwest::Client,
 	pub federation: reqwest::Client,
 	pub federation_slow: reqwest::Client,
 	pub sender: reqwest::Client,
@@ -43,7 +42,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 
 		Ok(Arc::new(Self {
 			default: base(config)?
-				.dns_resolver(resolver.resolver.clone())
+				.dns_resolver(resolver.dns.resolver.clone())
 				.build()?,
 
 			url_preview: base(config)
@@ -51,28 +50,19 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 					builder_interface(builder, url_preview_bind_iface.as_deref())
 				})?
 				.local_address(url_preview_bind_addr)
-				.dns_resolver(resolver.resolver.clone())
+				.dns_resolver(resolver.dns.resolver.clone())
 				.timeout(Duration::from_secs(config.url_preview_timeout))
 				.redirect(redirect::Policy::limited(3))
 				.user_agent(url_preview_user_agent)
 				.build()?,
 
 			extern_media: base(config)?
-				.dns_resolver(resolver.resolver.clone())
+				.dns_resolver(resolver.dns.resolver.clone())
 				.redirect(redirect::Policy::limited(3))
 				.build()?,
 
-			well_known: base(config)?
-				.dns_resolver(resolver.resolver.clone())
-				.connect_timeout(Duration::from_secs(config.well_known_conn_timeout))
-				.read_timeout(Duration::from_secs(config.well_known_timeout))
-				.timeout(Duration::from_secs(config.well_known_timeout))
-				.pool_max_idle_per_host(0)
-				.redirect(redirect::Policy::limited(4))
-				.build()?,
-
 			federation: base(config)?
-				.dns_resolver(resolver.resolver.hooked.clone())
+				.dns_resolver(resolver.dns.resolver.hooked.clone())
 				.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
 				.read_timeout(Duration::from_secs(config.federation_timeout))
 				.timeout(Duration::from_secs(
@@ -86,7 +76,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				.build()?,
 
 			federation_slow: base(config)?
-				.dns_resolver(resolver.resolver.hooked.clone())
+				.dns_resolver(resolver.dns.resolver.hooked.clone())
 				.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
 				.read_timeout(Duration::from_secs(config.federation_timeout.saturating_mul(6)))
 				.timeout(Duration::from_secs(
@@ -100,7 +90,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				.build()?,
 
 			sender: base(config)?
-				.dns_resolver(resolver.resolver.hooked.clone())
+				.dns_resolver(resolver.dns.resolver.hooked.clone())
 				.connect_timeout(Duration::from_secs(config.federation_conn_timeout))
 				.read_timeout(Duration::from_secs(config.sender_timeout))
 				.timeout(Duration::from_secs(config.sender_timeout))
@@ -110,7 +100,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				.build()?,
 
 			appservice: base(config)?
-				.dns_resolver(resolver.resolver.clone())
+				.dns_resolver(resolver.dns.resolver.clone())
 				.connect_timeout(Duration::from_secs(5))
 				.read_timeout(Duration::from_secs(config.appservice_timeout))
 				.timeout(Duration::from_secs(config.appservice_timeout))
@@ -120,7 +110,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				.build()?,
 
 			pusher: base(config)?
-				.dns_resolver(resolver.resolver.clone())
+				.dns_resolver(resolver.dns.resolver.clone())
 				.connect_timeout(Duration::from_secs(config.pusher_conn_timeout))
 				.timeout(Duration::from_secs(config.pusher_timeout))
 				.pool_max_idle_per_host(1)
@@ -151,7 +141,7 @@ pub fn valid_cidr_range(&self, ip: &IPAddress) -> bool {
 	}
 }
 
-fn base(config: &Config) -> Result<reqwest::ClientBuilder> {
+pub fn base(config: &Config) -> Result<reqwest::ClientBuilder> {
 	let mut builder = reqwest::Client::builder()
 		.hickory_dns(true)
 		.connect_timeout(Duration::from_secs(config.request_conn_timeout))

src/service/federation/execute.rs

@@ -7,6 +7,7 @@
 };
 use ipaddress::IPAddress;
 use reqwest::{Client, Method, Request, Response, Url};
+use resolvematrix::resolution::Resolution;
 use ruma::{
 	ServerName,
 	api::{
@@ -18,7 +19,7 @@
 	},
 };
 
-use crate::{SUPPORTED_VERSIONS, resolver::actual::ActualDest};
+use crate::SUPPORTED_VERSIONS;
 
 impl super::Service {
 	/// Sends a signed request to a remote server over federation.
@@ -130,10 +131,15 @@ pub async fn execute_on<'i, T, PathBuilderInput>(
 			))));
 		}
 
-		let actual = self.services.resolver.get_actual_dest(dest).await?;
+		let actual = self
+			.services
+			.resolver
+			.resolver
+			.resolve_server(dest.as_str())
+			.await?;
 
 		let request = Request::try_from(request.try_into_http_request::<Vec<u8>>(
-			actual.string().as_str(),
+			actual.base_url().as_str(),
 			authentication,
 			PathBuilderInput::create(),
 		)?)?;
@@ -150,7 +156,7 @@ pub async fn execute_on<'i, T, PathBuilderInput>(
 	async fn perform<T>(
 		&self,
 		dest: &ServerName,
-		actual: &ActualDest,
+		actual: &Resolution,
 		request: Request,
 		client: &Client,
 	) -> Result<T::IncomingResponse>
@@ -186,7 +192,9 @@ fn validate_url(&self, url: &Url) -> Result<()> {
 		if let Some(url_host) = url.host_str() {
 			if let Ok(ip) = IPAddress::parse(url_host) {
 				trace!("Checking request URL IP {ip:?}");
-				self.services.resolver.validate_ip(&ip)?;
+				if !self.services.client.valid_cidr_range(&ip) {
+					return Err!(BadServerResponse("Not allowed to send requests to this IP"));
+				}
 			}
 		}
 
@@ -198,7 +206,7 @@ fn validate_url(&self, url: &Url) -> Result<()> {
 	async fn handle_response<T>(
 		&self,
 		dest: &ServerName,
-		actual: &ActualDest,
+		actual: &Resolution,
 		method: &Method,
 		url: &Url,
 		response: Response,
@@ -231,7 +239,7 @@ async fn handle_response<T>(
 
 async fn into_http_response(
 	dest: &ServerName,
-	actual: &ActualDest,
+	actual: &Resolution,
 	method: &Method,
 	url: &Url,
 	mut response: Response,
@@ -243,7 +251,7 @@ async fn into_http_response(
 		request_url = %url,
 		response_url = %response.url(),
 		"Received response from {}",
-		actual.string(),
+		actual.base_url(),
 	);
 
 	let mut http_response_builder = http::Response::builder()
@@ -280,7 +288,7 @@ async fn into_http_response(
 }
 
 fn handle_error(
-	actual: &ActualDest,
+	actual: &Resolution,
 	method: &Method,
 	url: &Url,
 	mut e: reqwest::Error,

src/service/resolver/actual.rs

@@ -1,398 +0,0 @@
-use std::{
-	fmt::Debug,
-	net::{IpAddr, SocketAddr},
-};
-
-use conduwuit::{Err, Result, debug, debug_info, err, error, trace};
-use futures::{FutureExt, TryFutureExt};
-use hickory_resolver::{
-	net::{DnsError, NetError},
-	proto::rr::rdata::SRV,
-};
-use ipaddress::IPAddress;
-use ruma::ServerName;
-
-use super::{
-	cache::{CachedDest, CachedOverride, MAX_IPS},
-	fed::{FedDest, PortString, add_port_to_hostname, get_ip_with_port},
-};
-
-#[derive(Clone, Debug)]
-pub(crate) struct ActualDest {
-	pub(crate) dest: FedDest,
-	pub(crate) host: String,
-}
-
-impl ActualDest {
-	#[inline]
-	pub(crate) fn string(&self) -> String { self.dest.https_string() }
-}
-
-impl super::Service {
-	#[tracing::instrument(skip_all, level = "debug", name = "resolve")]
-	pub(crate) async fn get_actual_dest(&self, server_name: &ServerName) -> Result<ActualDest> {
-		let (CachedDest { dest, host, .. }, _cached) =
-			self.lookup_actual_dest(server_name).await?;
-
-		Ok(ActualDest { dest, host })
-	}
-
-	pub(crate) async fn lookup_actual_dest(
-		&self,
-		server_name: &ServerName,
-	) -> Result<(CachedDest, bool)> {
-		if let Ok(result) = self.cache.get_destination(server_name).await {
-			return Ok((result, true));
-		}
-
-		let _dedup = self.resolving.lock(server_name.as_str());
-		if let Ok(result) = self.cache.get_destination(server_name).await {
-			return Ok((result, true));
-		}
-
-		self.resolve_actual_dest(server_name, true)
-			.inspect_ok(|result| self.cache.set_destination(server_name, result))
-			.map_ok(|result| (result, false))
-			.boxed()
-			.await
-	}
-
-	/// Returns: `actual_destination`, host header
-	/// Implemented according to the specification at <https://matrix.org/docs/spec/server_server/r0.1.4#resolving-server-names>
-	/// Numbers in comments below refer to bullet points in linked section of
-	/// specification
-	#[tracing::instrument(name = "actual", level = "debug", skip(self, cache))]
-	pub async fn resolve_actual_dest(
-		&self,
-		dest: &ServerName,
-		cache: bool,
-	) -> Result<CachedDest> {
-		self.validate_dest(dest)?;
-		let mut host = dest.as_str().to_owned();
-		let actual_dest = match get_ip_with_port(dest.as_str()) {
-			| Some(host_port) => Self::actual_dest_1(host_port)?,
-			| None =>
-				if let Some(pos) = dest.as_str().find(':') {
-					self.actual_dest_2(dest, cache, pos).await?
-				} else {
-					self.services.server.check_running()?;
-					match self.request_well_known(dest.as_str()).await? {
-						| Some(delegated) =>
-							self.actual_dest_3(&mut host, cache, delegated).await?,
-						| _ => match self.query_srv_record(dest.as_str()).await? {
-							| Some(overrider) =>
-								self.actual_dest_4(&host, cache, overrider).await?,
-							| _ => self.actual_dest_5(dest, cache).await?,
-						},
-					}
-				},
-		};
-
-		// Can't use get_ip_with_port here because we don't want to add a port
-		// to an IP address if it wasn't specified
-		let host = if let Ok(addr) = host.parse::<SocketAddr>() {
-			FedDest::Literal(addr)
-		} else if let Ok(addr) = host.parse::<IpAddr>() {
-			FedDest::Named(addr.to_string(), FedDest::default_port())
-		} else if let Some(pos) = host.find(':') {
-			let (host, port) = host.split_at(pos);
-			FedDest::Named(
-				host.to_owned(),
-				port.try_into().unwrap_or_else(|_| FedDest::default_port()),
-			)
-		} else {
-			FedDest::Named(host, FedDest::default_port())
-		};
-
-		debug!("Actual destination: {actual_dest:?} hostname: {host:?}");
-		Ok(CachedDest {
-			dest: actual_dest,
-			host: host.uri_string(),
-			expire: CachedDest::default_expire(),
-		})
-	}
-
-	fn actual_dest_1(host_port: FedDest) -> Result<FedDest> {
-		debug!("1: IP literal with provided or default port");
-		Ok(host_port)
-	}
-
-	async fn actual_dest_2(&self, dest: &ServerName, cache: bool, pos: usize) -> Result<FedDest> {
-		debug!("2: Hostname with included port");
-		let (host, port) = dest.as_str().split_at(pos);
-		self.conditional_query_and_cache(host, port.parse::<u16>().unwrap_or(8448), cache)
-			.await?;
-
-		Ok(FedDest::Named(
-			host.to_owned(),
-			port.try_into().unwrap_or_else(|_| FedDest::default_port()),
-		))
-	}
-
-	async fn actual_dest_3(
-		&self,
-		host: &mut String,
-		cache: bool,
-		delegated: String,
-	) -> Result<FedDest> {
-		debug!("3: A .well-known file is available");
-		*host = add_port_to_hostname(&delegated).uri_string();
-		match get_ip_with_port(&delegated) {
-			| Some(host_and_port) => Self::actual_dest_3_1(host_and_port),
-			| None =>
-				if let Some(pos) = delegated.find(':') {
-					self.actual_dest_3_2(cache, delegated, pos).await
-				} else {
-					trace!("Delegated hostname has no port in this branch");
-					match self.query_srv_record(&delegated).await? {
-						| Some(overrider) =>
-							self.actual_dest_3_3(cache, delegated, overrider).await,
-						| _ => self.actual_dest_3_4(cache, delegated).await,
-					}
-				},
-		}
-	}
-
-	fn actual_dest_3_1(host_and_port: FedDest) -> Result<FedDest> {
-		debug!("3.1: IP literal in .well-known file");
-		Ok(host_and_port)
-	}
-
-	async fn actual_dest_3_2(
-		&self,
-		cache: bool,
-		delegated: String,
-		pos: usize,
-	) -> Result<FedDest> {
-		debug!("3.2: Hostname with port in .well-known file");
-		let (host, port) = delegated.split_at(pos);
-		self.conditional_query_and_cache(host, port.parse::<u16>().unwrap_or(8448), cache)
-			.await?;
-
-		Ok(FedDest::Named(
-			host.to_owned(),
-			port.try_into().unwrap_or_else(|_| FedDest::default_port()),
-		))
-	}
-
-	async fn actual_dest_3_3(
-		&self,
-		cache: bool,
-		delegated: String,
-		overrider: FedDest,
-	) -> Result<FedDest> {
-		debug!("3.3: SRV lookup successful");
-		let force_port = overrider.port();
-		self.conditional_query_and_cache_override(
-			&delegated,
-			&overrider.hostname(),
-			force_port.unwrap_or(8448),
-			cache,
-		)
-		.await?;
-
-		if let Some(port) = force_port {
-			return Ok(FedDest::Named(
-				delegated,
-				format!(":{port}")
-					.as_str()
-					.try_into()
-					.unwrap_or_else(|_| FedDest::default_port()),
-			));
-		}
-
-		Ok(add_port_to_hostname(&delegated))
-	}
-
-	async fn actual_dest_3_4(&self, cache: bool, delegated: String) -> Result<FedDest> {
-		debug!("3.4: No SRV records, just use the hostname from .well-known");
-		self.conditional_query_and_cache(&delegated, 8448, cache)
-			.await?;
-		Ok(add_port_to_hostname(&delegated))
-	}
-
-	async fn actual_dest_4(
-		&self,
-		host: &str,
-		cache: bool,
-		overrider: FedDest,
-	) -> Result<FedDest> {
-		debug!("4: No .well-known; SRV record found");
-		let force_port = overrider.port();
-		self.conditional_query_and_cache_override(
-			host,
-			&overrider.hostname(),
-			force_port.unwrap_or(8448),
-			cache,
-		)
-		.await?;
-
-		if let Some(port) = force_port {
-			let port = format!(":{port}");
-
-			return Ok(FedDest::Named(
-				host.to_owned(),
-				PortString::from(port.as_str()).unwrap_or_else(|_| FedDest::default_port()),
-			));
-		}
-
-		Ok(add_port_to_hostname(host))
-	}
-
-	async fn actual_dest_5(&self, dest: &ServerName, cache: bool) -> Result<FedDest> {
-		debug!("5: No SRV record found");
-		self.conditional_query_and_cache(dest.as_str(), 8448, cache)
-			.await?;
-
-		Ok(add_port_to_hostname(dest.as_str()))
-	}
-
-	#[inline]
-	async fn conditional_query_and_cache(
-		&self,
-		hostname: &str,
-		port: u16,
-		cache: bool,
-	) -> Result {
-		self.conditional_query_and_cache_override(hostname, hostname, port, cache)
-			.await
-	}
-
-	#[inline]
-	async fn conditional_query_and_cache_override(
-		&self,
-		untername: &str,
-		hostname: &str,
-		port: u16,
-		cache: bool,
-	) -> Result {
-		if !cache {
-			return Ok(());
-		}
-
-		if self.cache.has_override(untername).await {
-			return Ok(());
-		}
-
-		self.query_and_cache_override(untername, hostname, port)
-			.await
-	}
-
-	#[tracing::instrument(name = "ip", level = "debug", skip(self))]
-	async fn query_and_cache_override(
-		&self,
-		untername: &'_ str,
-		hostname: &'_ str,
-		port: u16,
-	) -> Result {
-		self.services.server.check_running()?;
-
-		debug!("querying IP for {untername:?} ({hostname:?}:{port})");
-		match self.resolver.resolver.lookup_ip(hostname.to_owned()).await {
-			| Err(e) => Self::handle_resolve_error(&e, hostname),
-			| Ok(override_ip) => {
-				self.cache.set_override(untername, &CachedOverride {
-					ips: override_ip.iter().take(MAX_IPS).collect(),
-					port,
-					expire: CachedOverride::default_expire(),
-					overriding: (hostname != untername)
-						.then_some(hostname.into())
-						.inspect(|_| debug_info!("{untername:?} overridden by {hostname:?}")),
-				});
-
-				Ok(())
-			},
-		}
-	}
-
-	#[tracing::instrument(name = "srv", level = "debug", skip(self))]
-	async fn query_srv_record(&self, hostname: &'_ str) -> Result<Option<FedDest>> {
-		let hostnames =
-			[format!("_matrix-fed._tcp.{hostname}."), format!("_matrix._tcp.{hostname}.")];
-
-		for hostname in hostnames {
-			self.services.server.check_running()?;
-
-			debug!("querying SRV for {hostname:?}");
-			let hostname = hostname.trim_end_matches('.');
-			match self.resolver.resolver.srv_lookup(hostname).await {
-				| Err(e) => Self::handle_resolve_error(&e, hostname)?,
-				| Ok(result) => {
-					return Ok(result.answers().iter().next().map(|result| {
-						let data = result.try_borrow::<SRV>().expect("should be SRV response");
-
-						FedDest::Named(
-							data.data()
-								.target
-								.to_string()
-								.trim_end_matches('.')
-								.to_owned(),
-							format!(":{}", data.data().port)
-								.as_str()
-								.try_into()
-								.unwrap_or_else(|_| FedDest::default_port()),
-						)
-					}));
-				},
-			}
-		}
-
-		Ok(None)
-	}
-
-	fn handle_resolve_error(err: &NetError, host: &'_ str) -> Result<()> {
-		match err {
-			| NetError::NoConnections => {
-				error!(
-					"Your DNS server is overloaded and has ran out of connections. It is \
-					 strongly recommended you remediate this issue to ensure proper federation \
-					 connectivity."
-				);
-
-				Err!(error!(%host, "DNS error: {err}"))
-			},
-			| NetError::Timeout => Err!(error!(%host, "DNS query timed out")),
-			| NetError::Dns(DnsError::NoRecordsFound(..)) => {
-				// Raise to debug_warn if we can find out the result wasn't from cache
-				debug!(%host, "No DNS records found: {err}");
-				Ok(())
-			},
-			| _ => Err!(error!(%host, "DNS error: {err}")),
-		}
-	}
-
-	fn validate_dest(&self, dest: &ServerName) -> Result<()> {
-		if dest == self.services.server.name && !self.services.server.config.federation_loopback {
-			return Err!("Won't send federation request to ourselves");
-		}
-
-		if dest.is_ip_literal() || IPAddress::is_valid(dest.host()) {
-			self.validate_dest_ip_literal(dest)?;
-		}
-
-		Ok(())
-	}
-
-	fn validate_dest_ip_literal(&self, dest: &ServerName) -> Result<()> {
-		trace!("Destination is an IP literal, checking against IP range denylist.",);
-		debug_assert!(
-			dest.is_ip_literal() || !IPAddress::is_valid(dest.host()),
-			"Destination is not an IP literal."
-		);
-		let ip = IPAddress::parse(dest.host()).map_err(|e| {
-			err!(BadServerResponse(debug_error!("Failed to parse IP literal from string: {e}")))
-		})?;
-
-		self.validate_ip(&ip)?;
-
-		Ok(())
-	}
-
-	pub(crate) fn validate_ip(&self, ip: &IPAddress) -> Result<()> {
-		if !self.services.client.valid_cidr_range(ip) {
-			return Err!(BadServerResponse("Not allowed to send requests to this IP"));
-		}
-
-		Ok(())
-	}
-}

src/service/resolver/cache.rs

@@ -4,7 +4,7 @@
 	Result,
 	arrayvec::ArrayVec,
 	at, err,
-	utils::{math::Expected, rand, stream::TryIgnore},
+	utils::{math::Expected, stream::TryIgnore},
 };
 use database::{Cbor, Deserialized, Map};
 use futures::{Stream, StreamExt, future::join};
@@ -114,11 +114,6 @@ impl CachedDest {
 	#[must_use]
 	pub fn valid(&self) -> bool { self.expire > SystemTime::now() }
 
-	#[must_use]
-	pub(crate) fn default_expire() -> SystemTime {
-		rand::time_from_now_secs(60 * 60 * 18..60 * 60 * 36)
-	}
-
 	#[inline]
 	#[must_use]
 	pub fn size(&self) -> usize {
@@ -134,11 +129,6 @@ impl CachedOverride {
 	#[must_use]
 	pub fn valid(&self) -> bool { self.expire > SystemTime::now() }
 
-	#[must_use]
-	pub(crate) fn default_expire() -> SystemTime {
-		rand::time_from_now_secs(60 * 60 * 6..60 * 60 * 12)
-	}
-
 	#[inline]
 	#[must_use]
 	pub fn size(&self) -> usize { size_of_val(self) }

src/service/resolver/fed.rs

@@ -1,16 +1,12 @@
-use std::{
-	borrow::Cow,
-	fmt,
-	net::{IpAddr, SocketAddr},
-};
+use std::{fmt, net::SocketAddr};
 
 use conduwuit::{arrayvec::ArrayString, utils::math::Expected};
 use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 pub enum FedDest {
-	Literal(SocketAddr),
-	Named(String, PortString),
+	Literal(SocketAddr),       // "ip:port"
+	Named(String, PortString), // ("hostname", ":port")
 }
 
 /// numeric or service-name
@@ -18,36 +14,7 @@ pub enum FedDest {
 
 const DEFAULT_PORT: &str = ":8448";
 
-pub(crate) fn get_ip_with_port(dest_str: &str) -> Option<FedDest> {
-	if let Ok(dest) = dest_str.parse::<SocketAddr>() {
-		Some(FedDest::Literal(dest))
-	} else if let Ok(ip_addr) = dest_str.parse::<IpAddr>() {
-		Some(FedDest::Literal(SocketAddr::new(ip_addr, 8448)))
-	} else {
-		None
-	}
-}
-
-pub(crate) fn add_port_to_hostname(dest: &str) -> FedDest {
-	let (host, port) = match dest.find(':') {
-		| None => (dest, DEFAULT_PORT),
-		| Some(pos) => dest.split_at(pos),
-	};
-
-	FedDest::Named(
-		host.to_owned(),
-		PortString::from(port).unwrap_or_else(|_| FedDest::default_port()),
-	)
-}
-
 impl FedDest {
-	pub(crate) fn https_string(&self) -> String {
-		match self {
-			| Self::Literal(addr) => format!("https://{addr}"),
-			| Self::Named(host, port) => format!("https://{host}{port}"),
-		}
-	}
-
 	pub(crate) fn uri_string(&self) -> String {
 		match self {
 			| Self::Literal(addr) => addr.to_string(),
@@ -55,23 +22,6 @@ pub(crate) fn uri_string(&self) -> String {
 		}
 	}
 
-	#[inline]
-	pub(crate) fn hostname(&self) -> Cow<'_, str> {
-		match &self {
-			| Self::Literal(addr) => addr.ip().to_string().into(),
-			| Self::Named(host, _) => host.into(),
-		}
-	}
-
-	#[inline]
-	#[allow(clippy::string_slice)]
-	pub(crate) fn port(&self) -> Option<u16> {
-		match &self {
-			| Self::Literal(addr) => Some(addr.port()),
-			| Self::Named(_, port) => port[1..].parse().ok(),
-		}
-	}
-
 	#[inline]
 	#[must_use]
 	pub fn default_port() -> PortString {

src/service/resolver/mod.rs

@@ -1,54 +1,64 @@
-pub mod actual;
 pub mod cache;
 mod dns;
 pub mod fed;
-#[cfg(test)]
-mod tests;
-mod well_known;
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use async_trait::async_trait;
-use conduwuit::{Result, Server, arrayvec::ArrayString, utils::MutexMap};
+use conduwuit::Result;
+use reqwest::redirect;
+use resolvematrix::server::{MatrixResolver, MatrixResolverBuilder};
 
 use self::{cache::Cache, dns::Resolver};
-use crate::{Dep, client};
+use crate::client::base;
 
 pub struct Service {
-	pub cache: Arc<Cache>,
-	pub resolver: Arc<Resolver>,
-	resolving: Resolving,
+	pub resolver: MatrixResolver,
+	pub dns: Dns,
+	#[allow(dead_code)] // This service doesn't access services after construction
 	services: Services,
 }
 
-struct Services {
-	server: Arc<Server>,
-	client: Dep<client::Service>,
-}
+struct Services;
 
-type Resolving = MutexMap<NameBuf, ()>;
-type NameBuf = ArrayString<256>;
+pub struct Dns {
+	pub cache: Arc<Cache>,
+	pub resolver: Arc<Resolver>,
+}
 
 #[async_trait]
 impl crate::Service for Service {
 	#[allow(clippy::as_conversions, clippy::cast_sign_loss, clippy::cast_possible_truncation)]
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		let cache = Cache::new(&args);
+		let resolver = Resolver::build(args.server, cache.clone())?;
 		Ok(Arc::new(Self {
-			cache: cache.clone(),
-			resolver: Resolver::build(args.server, cache)?,
-			resolving: MutexMap::new(),
-			services: Services {
-				server: args.server.clone(),
-				client: args.depend::<client::Service>("client"),
+			resolver: MatrixResolverBuilder::new()
+				.dangerous_tls_accept_invalid_certs(args.server.config.allow_invalid_tls_certificates_yes_i_know_what_the_fuck_i_am_doing_with_this_and_i_know_this_is_insecure)
+				.http_client(
+					base(&args.server.config)?
+						.connect_timeout(Duration::from_secs(args.server.config.well_known_conn_timeout))
+						.read_timeout(Duration::from_secs(args.server.config.well_known_timeout))
+						.timeout(Duration::from_secs(args.server.config.well_known_timeout))
+						.pool_max_idle_per_host(0)
+						.redirect(redirect::Policy::limited(4))
+						.build()?
+				)
+				.dns_resolver(resolver.resolver.clone())
+				.build()?,
+			dns: Dns {
+				cache,
+				resolver,
 			},
+			services: Services {},
 		}))
 	}
 
 	async fn clear_cache(&self) {
 		self.resolver.clear_cache();
-		self.cache.clear().await;
+		self.dns.resolver.clear_cache();
+		self.dns.cache.clear().await;
 	}
 
-	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
+	fn name(&self) -> &str { crate::service::make_name(module_path!()) }
 }

src/service/resolver/tests.rs

@@ -1,41 +0,0 @@
-use super::fed::{FedDest, add_port_to_hostname, get_ip_with_port};
-
-#[test]
-fn ips_get_default_ports() {
-	assert_eq!(
-		get_ip_with_port("1.1.1.1"),
-		Some(FedDest::Literal("1.1.1.1:8448".parse().unwrap()))
-	);
-	assert_eq!(
-		get_ip_with_port("dead:beef::"),
-		Some(FedDest::Literal("[dead:beef::]:8448".parse().unwrap()))
-	);
-}
-
-#[test]
-fn ips_keep_custom_ports() {
-	assert_eq!(
-		get_ip_with_port("1.1.1.1:1234"),
-		Some(FedDest::Literal("1.1.1.1:1234".parse().unwrap()))
-	);
-	assert_eq!(
-		get_ip_with_port("[dead::beef]:8933"),
-		Some(FedDest::Literal("[dead::beef]:8933".parse().unwrap()))
-	);
-}
-
-#[test]
-fn hostnames_get_default_ports() {
-	assert_eq!(
-		add_port_to_hostname("example.com"),
-		FedDest::Named(String::from("example.com"), ":8448".try_into().unwrap())
-	);
-}
-
-#[test]
-fn hostnames_keep_custom_ports() {
-	assert_eq!(
-		add_port_to_hostname("example.com:1337"),
-		FedDest::Named(String::from("example.com"), ":1337".try_into().unwrap())
-	);
-}

src/service/resolver/well_known.rs

@@ -1,50 +0,0 @@
-use conduwuit::{Result, debug, debug_error, debug_info, trace, utils::response::LimitReadExt};
-use ruma::ServerName;
-
-impl super::Service {
-	#[tracing::instrument(name = "well-known", level = "debug", skip(self, dest))]
-	pub(super) async fn request_well_known(&self, dest: &str) -> Result<Option<String>> {
-		trace!("Requesting well known for {dest}");
-		let response = self
-			.services
-			.client
-			.well_known
-			.get(format!("https://{dest}/.well-known/matrix/server"))
-			.send()
-			.await;
-
-		trace!("response: {response:?}");
-		if let Err(e) = &response {
-			debug!("error: {e:?}");
-			return Ok(None);
-		}
-
-		let response = response?;
-		if !response.status().is_success() {
-			debug!("response not 2XX");
-			return Ok(None);
-		}
-
-		let Ok(text) = response.limit_read_text(8192).await else {
-			debug!("failed to read well-known response (too large or non-text content)");
-			return Ok(None);
-		};
-		trace!("response text: {text:?}");
-
-		let body: serde_json::Value = serde_json::from_str(&text).unwrap_or_default();
-
-		let m_server = body
-			.get("m.server")
-			.unwrap_or(&serde_json::Value::Null)
-			.as_str()
-			.unwrap_or_default();
-
-		if ServerName::parse(m_server).is_err() {
-			debug_error!("response content missing or invalid");
-			return Ok(None);
-		}
-
-		debug_info!("{dest:?} found at {m_server:?}");
-		Ok(Some(m_server.to_owned()))
-	}
-}

src/service/rooms/event_handler/handle_incoming_pdu.rs

@@ -194,10 +194,14 @@ pub async fn handle_incoming_pdu<'a>(
 						"Invite to {room_id} appears to have been rescinded by {sender}, \
 						 marking as left"
 					);
+
 					self.services
 						.state_cache
 						.mark_as_left(&sender, room_id, Some(pdu))
 						.await;
+
+					self.services.sync.wake(&sender).await;
+
 					return Ok(None);
 				}
 			}

src/service/rooms/event_handler/mod.rs

@@ -21,7 +21,7 @@
 };
 use tokio::sync::{Notify, mpsc};
 
-use crate::{Dep, globals, rooms, sending, server_keys};
+use crate::{Dep, globals, rooms, sending, server_keys, sync};
 pub struct Service {
 	pub mutex_federation: RoomMutexMap,
 	pub federation_handletime: SyncRwLock<HandleTimeMap>,
@@ -44,6 +44,7 @@ struct Services {
 	state_cache: Dep<rooms::state_cache::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
 	state_compressor: Dep<rooms::state_compressor::Service>,
+	sync: Dep<sync::Service>,
 	timeline: Dep<rooms::timeline::Service>,
 	server: Arc<Server>,
 }
@@ -74,6 +75,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
 				state_compressor: args
 					.depend::<rooms::state_compressor::Service>("rooms::state_compressor"),
+				sync: args.depend::<sync::Service>("sync"),
 				timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
 				server: args.server.clone(),
 			},

src/service/rooms/membership/mod.rs

@@ -40,7 +40,7 @@
 		state_compressor::{self, CompressedState, HashSetCompressStateEvent},
 		timeline::{self, pdu_fits},
 	},
-	sending, server_keys, users,
+	sending, server_keys, sync, users,
 };
 
 pub struct Service {
@@ -63,6 +63,7 @@ struct Services {
 	state_accessor: Dep<state_accessor::Service>,
 	state_cache: Dep<state_cache::Service>,
 	state_compressor: Dep<state_compressor::Service>,
+	sync: Dep<sync::Service>,
 	timeline: Dep<timeline::Service>,
 	users: Dep<users::Service>,
 }
@@ -87,6 +88,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				state_cache: args.depend::<state_cache::Service>("rooms::state_cache"),
 				state_compressor: args
 					.depend::<state_compressor::Service>("rooms::state_compressor"),
+				sync: args.depend::<sync::Service>("sync"),
 				timeline: args.depend::<timeline::Service>("rooms::timeline"),
 				users: args.depend::<users::Service>("users"),
 			},
@@ -672,6 +674,8 @@ pub async fn join_remote_room(
 		}
 		drop(cork);
 
+		self.services.sync.wake_all_joined(room_id).await;
+
 		Ok(())
 	}
 

src/service/rooms/read_receipt/mod.rs

@@ -21,7 +21,7 @@
 };
 
 use self::data::{Data, ReceiptItem};
-use crate::{Dep, rooms, sending};
+use crate::{Dep, rooms, sending, sync};
 
 pub struct Service {
 	services: Services,
@@ -31,6 +31,7 @@ pub struct Service {
 struct Services {
 	sending: Dep<sending::Service>,
 	short: Dep<rooms::short::Service>,
+	sync: Dep<sync::Service>,
 	timeline: Dep<rooms::timeline::Service>,
 }
 
@@ -40,6 +41,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 			services: Services {
 				sending: args.depend::<sending::Service>("sending"),
 				short: args.depend::<rooms::short::Service>("rooms::short"),
+				sync: args.depend::<sync::Service>("sync"),
 				timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
 			},
 			db: Data::new(&args),
@@ -63,6 +65,7 @@ pub async fn readreceipt_update(
 			.flush_room(room_id)
 			.await
 			.expect("room flush failed");
+		self.services.sync.wake_all_joined(room_id).await;
 	}
 
 	/// Gets the latest private read receipt from the user in the room

src/service/rooms/state/mod.rs

@@ -24,11 +24,13 @@
 };
 
 use crate::{
-	Dep, globals, rooms,
+	Dep, globals,
 	rooms::{
+		self,
 		short::{ShortEventId, ShortStateHash},
 		state_compressor::{CompressedState, parse_compressed_state_event},
 	},
+	sync,
 };
 
 pub struct Service {
@@ -43,6 +45,7 @@ struct Services {
 	state_cache: Dep<rooms::state_cache::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
 	state_compressor: Dep<rooms::state_compressor::Service>,
+	sync: Dep<sync::Service>,
 	timeline: Dep<rooms::timeline::Service>,
 }
 
@@ -68,6 +71,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
 				state_compressor: args
 					.depend::<rooms::state_compressor::Service>("rooms::state_compressor"),
+				sync: args.depend::<sync::Service>("sync"),
 				timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
 			},
 			db: Data {
@@ -135,6 +139,8 @@ pub async fn force_state(
 
 		self.set_room_state(room_id, shortstatehash, state_lock);
 
+		self.services.sync.wake_all_joined(room_id).await;
+
 		Ok(())
 	}
 
@@ -296,23 +302,19 @@ pub async fn append_to_state(&self, new_pdu: &PduEvent, room_id: &RoomId) -> Res
 	}
 
 	#[tracing::instrument(skip_all, level = "debug")]
-	pub async fn summary_stripped<'a, E>(
+	pub async fn summary_stripped(
 		&self,
-		event: &'a E,
+		event: &PduEvent,
 		room_id: &RoomId,
 		target_user: &UserId,
-	) -> Vec<RawStrippedState>
-	where
-		E: Event + Send + Sync,
-		&'a E: Event + Send,
-	{
+	) -> Vec<RawStrippedState> {
 		let mut state_events = [
 			(&StateEventType::RoomCreate, ""),
 			(&StateEventType::RoomJoinRules, ""),
 			(&StateEventType::RoomCanonicalAlias, ""),
 			(&StateEventType::RoomName, ""),
 			(&StateEventType::RoomAvatar, ""),
-			(&StateEventType::RoomMember, event.sender().as_str()), // Add recommended events
+			(&StateEventType::RoomMember, event.sender().as_str()),
 			(&StateEventType::RoomEncryption, ""),
 			(&StateEventType::RoomTopic, ""),
 		]
@@ -322,11 +324,20 @@ pub async fn summary_stripped<'a, E>(
 			state_events.push((&StateEventType::RoomMember, target_user.as_str()));
 		}
 
-		let fetches = state_events.into_iter().map(|(event_type, state_key)| {
-			self.services
-				.state_accessor
-				.room_state_get(room_id, event_type, state_key)
-		});
+		let fetches = state_events
+			.into_iter()
+			.map(async |(event_type, state_key)| {
+				if event.event_type() == &TimelineEventType::from(event_type.clone())
+					&& event.state_key() == Some(state_key)
+				{
+					Ok(event.clone())
+				} else {
+					self.services
+						.state_accessor
+						.room_state_get(room_id, event_type, state_key)
+						.await
+				}
+			});
 
 		join_all(fetches)
 			.await

src/service/rooms/state_cache/mod.rs

@@ -16,7 +16,9 @@
 	serde::Raw,
 };
 
-use crate::{Dep, account_data, appservice::RegistrationInfo, config, globals, rooms, users};
+use crate::{
+	Dep, account_data, appservice::RegistrationInfo, config, globals, rooms, sync, users,
+};
 
 pub struct Service {
 	appservice_in_room_cache: AppServiceInRoomCache,
@@ -31,6 +33,7 @@ struct Services {
 	metadata: Dep<rooms::metadata::Service>,
 	state: Dep<rooms::state::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
+	sync: Dep<sync::Service>,
 	users: Dep<users::Service>,
 }
 
@@ -67,6 +70,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				state: args.depend::<rooms::state::Service>("rooms::state"),
 				state_accessor: args
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
+				sync: args.depend::<sync::Service>("sync"),
 				users: args.depend::<users::Service>("users"),
 			},
 			db: Data {

src/service/rooms/state_cache/update.rs

@@ -29,8 +29,9 @@ pub async fn update_membership(
 		update_joined_count: bool,
 	) -> Result {
 		let membership = pdu.get_content::<RoomMemberEventContent>()?;
+		let is_local = self.services.globals.user_is_local(user_id);
 
-		if !self.services.globals.user_is_local(user_id) {
+		if !is_local {
 			self.services.users.record_remote_user(user_id);
 		}
 
@@ -41,13 +42,14 @@ pub async fn update_membership(
 					// Add the user ID to the join list then
 					self.mark_as_once_joined(user_id, room_id);
 
-					// Check if the room has a predecessor
-					if let Ok(Some(predecessor)) = self
-						.services
-						.state_accessor
-						.room_state_get_content(room_id, &StateEventType::RoomCreate, "")
-						.await
-						.map(|content: RoomCreateEventContent| content.predecessor)
+					// Copy data from the predecessor if the user is local
+					if is_local
+						&& let Ok(Some(predecessor)) = self
+							.services
+							.state_accessor
+							.room_state_get_content(room_id, &StateEventType::RoomCreate, "")
+							.await
+							.map(|content: RoomCreateEventContent| content.predecessor)
 					{
 						// Copy old tags to new room
 						if let Ok(tag_event) = self
@@ -109,13 +111,16 @@ pub async fn update_membership(
 				self.mark_as_joined(user_id, room_id);
 			},
 			| MembershipState::Invite => {
-				let last_state = self
-					.services
-					.state
-					.summary_stripped(pdu, room_id, user_id)
-					.await;
+				let invite_state = if is_local {
+					self.services
+						.state
+						.summary_stripped(pdu, room_id, user_id)
+						.await
+				} else {
+					vec![]
+				};
 
-				self.mark_as_invited(user_id, room_id, pdu.sender(), last_state, None)
+				self.mark_as_invited(user_id, room_id, pdu.sender(), invite_state, None)
 					.await?;
 			},
 			| MembershipState::Leave | MembershipState::Ban => {
@@ -128,6 +133,13 @@ pub async fn update_membership(
 			self.update_joined_count(room_id).await;
 		}
 
+		// Kick the target user's sync loop if they're local and this isn't a join to
+		// make sure that membership changes like invites or invite rejections get
+		// synced
+		if is_local && !matches!(membership.membership, MembershipState::Join) {
+			self.services.sync.wake(user_id).await;
+		}
+
 		Ok(())
 	}
 

src/service/rooms/timeline/append.rs

@@ -86,6 +86,8 @@ pub async fn append_incoming_pdu<'a, Leaves>(
 			}
 		}
 
+		self.services.sync.wake_all_joined(room_id).await;
+
 		Ok(Some(pdu_id))
 	}
 
@@ -187,18 +189,6 @@ pub async fn append_pdu<'a, Leaves>(
 
 		let count1 = self.services.globals.next_count().unwrap();
 
-		// Mark as read first so the sending client doesn't get a notification even if
-		// appending fails
-		// TODO: Is this necessary? appending doesn't seem that fallible, and if it is,
-		// there's bigger issues than ghost notifications.
-		self.services
-			.read_receipt
-			.private_read_set(room_id, pdu.sender(), count1);
-
-		self.services
-			.user
-			.reset_notification_counts(pdu.sender(), room_id);
-
 		let count2 = PduCount::Normal(self.services.globals.next_count().unwrap());
 		let pdu_id: RawPduId = PduId { shortroomid, shorteventid: count2 }.into();
 
@@ -223,6 +213,14 @@ pub async fn append_pdu<'a, Leaves>(
 			);
 		}
 
+		self.services
+			.read_receipt
+			.private_read_set(room_id, pdu.sender(), count1);
+
+		self.services
+			.user
+			.reset_notification_counts(pdu.sender(), room_id);
+
 		self.send_to_interested_appservices(pdu, &pdu_id, room_id)
 			.await;
 

src/service/rooms/timeline/build.rs

@@ -153,6 +153,8 @@ pub async fn build_and_append_pdu(
 			.state
 			.set_room_state(&room_id, statehashid, state_lock);
 
+		self.services.sync.wake_all_joined(&room_id).await;
+
 		let mut servers: HashSet<OwnedServerName> = self
 			.services
 			.state_cache

src/service/rooms/timeline/mod.rs

@@ -28,7 +28,8 @@
 use self::data::Data;
 pub use self::{create::pdu_fits, data::PdusIterItem};
 use crate::{
-	Dep, account_data, admin, appservice, globals, pusher, rooms, sending, server_keys, users,
+	Dep, account_data, admin, appservice, globals, pusher, rooms, sending, server_keys, sync,
+	users,
 };
 
 // Update Relationships
@@ -65,21 +66,22 @@ struct Services {
 	appservice: Dep<appservice::Service>,
 	admin: Dep<admin::Service>,
 	alias: Dep<rooms::alias::Service>,
+	event_handler: Dep<rooms::event_handler::Service>,
 	globals: Dep<globals::Service>,
-	short: Dep<rooms::short::Service>,
-	state: Dep<rooms::state::Service>,
-	state_cache: Dep<rooms::state_cache::Service>,
-	state_accessor: Dep<rooms::state_accessor::Service>,
 	pdu_metadata: Dep<rooms::pdu_metadata::Service>,
+	pusher: Dep<pusher::Service>,
 	read_receipt: Dep<rooms::read_receipt::Service>,
+	search: Dep<rooms::search::Service>,
 	sending: Dep<sending::Service>,
 	server_keys: Dep<server_keys::Service>,
+	short: Dep<rooms::short::Service>,
+	state: Dep<rooms::state::Service>,
+	state_accessor: Dep<rooms::state_accessor::Service>,
+	state_cache: Dep<rooms::state_cache::Service>,
+	sync: Dep<sync::Service>,
+	threads: Dep<rooms::threads::Service>,
 	user: Dep<rooms::user::Service>,
 	users: Dep<users::Service>,
-	pusher: Dep<pusher::Service>,
-	threads: Dep<rooms::threads::Service>,
-	search: Dep<rooms::search::Service>,
-	event_handler: Dep<rooms::event_handler::Service>,
 }
 
 type RoomMutexMap = MutexMap<OwnedRoomId, ()>;
@@ -95,23 +97,24 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				appservice: args.depend::<appservice::Service>("appservice"),
 				admin: args.depend::<admin::Service>("admin"),
 				alias: args.depend::<rooms::alias::Service>("rooms::alias"),
-				globals: args.depend::<globals::Service>("globals"),
-				short: args.depend::<rooms::short::Service>("rooms::short"),
-				state: args.depend::<rooms::state::Service>("rooms::state"),
-				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
-				state_accessor: args
-					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
-				pdu_metadata: args.depend::<rooms::pdu_metadata::Service>("rooms::pdu_metadata"),
-				read_receipt: args.depend::<rooms::read_receipt::Service>("rooms::read_receipt"),
-				sending: args.depend::<sending::Service>("sending"),
-				server_keys: args.depend::<server_keys::Service>("server_keys"),
-				user: args.depend::<rooms::user::Service>("rooms::user"),
-				users: args.depend::<users::Service>("users"),
-				pusher: args.depend::<pusher::Service>("pusher"),
-				threads: args.depend::<rooms::threads::Service>("rooms::threads"),
-				search: args.depend::<rooms::search::Service>("rooms::search"),
 				event_handler: args
 					.depend::<rooms::event_handler::Service>("rooms::event_handler"),
+				globals: args.depend::<globals::Service>("globals"),
+				pdu_metadata: args.depend::<rooms::pdu_metadata::Service>("rooms::pdu_metadata"),
+				pusher: args.depend::<pusher::Service>("pusher"),
+				read_receipt: args.depend::<rooms::read_receipt::Service>("rooms::read_receipt"),
+				search: args.depend::<rooms::search::Service>("rooms::search"),
+				sending: args.depend::<sending::Service>("sending"),
+				server_keys: args.depend::<server_keys::Service>("server_keys"),
+				short: args.depend::<rooms::short::Service>("rooms::short"),
+				state: args.depend::<rooms::state::Service>("rooms::state"),
+				state_accessor: args
+					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
+				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
+				sync: args.depend::<sync::Service>("sync"),
+				threads: args.depend::<rooms::threads::Service>("rooms::threads"),
+				user: args.depend::<rooms::user::Service>("rooms::user"),
+				users: args.depend::<users::Service>("users"),
 			},
 			db: Data::new(&args),
 			mutex_insert: RoomMutexMap::new(),

src/service/sync/mod.rs

@@ -1,42 +1,26 @@
-mod watch;
-
 use std::{
-	collections::{BTreeMap, BTreeSet},
+	collections::{BTreeMap, BTreeSet, HashMap},
+	pin::pin,
 	sync::Arc,
 };
 
-use conduwuit::{Result, Server, SyncMutex};
-use database::Map;
-use ruma::{OwnedDeviceId, OwnedRoomId, OwnedUserId, api::client::sync::sync_events::v5};
+use conduwuit::{Result, SyncMutex, trace};
+use futures::StreamExt;
+use ruma::{
+	OwnedDeviceId, OwnedRoomId, OwnedUserId, RoomId, UserId, api::client::sync::sync_events::v5,
+};
+use tokio::sync::{Mutex, Notify};
 
 use crate::{Dep, rooms};
 
 pub struct Service {
-	db: Data,
 	services: Services,
-	connections: DbConnections<DbConnectionsKey, DbConnectionsVal>,
+	wakers: Mutex<HashMap<OwnedUserId, Arc<Notify>>>,
 	snake_connections: DbConnections<SnakeConnectionsKey, SnakeConnectionsVal>,
 }
 
-pub struct Data {
-	todeviceid_events: Arc<Map>,
-	userroomid_joined: Arc<Map>,
-	userroomid_invitestate: Arc<Map>,
-	userroomid_leftstate: Arc<Map>,
-	userroomid_notificationcount: Arc<Map>,
-	userroomid_highlightcount: Arc<Map>,
-	pduid_pdu: Arc<Map>,
-	keychangeid_userid: Arc<Map>,
-	roomusertype_roomuserdataid: Arc<Map>,
-	readreceiptid_readreceipt: Arc<Map>,
-	userid_lastonetimekeyupdate: Arc<Map>,
-}
-
 struct Services {
-	server: Arc<Server>,
-	short: Dep<rooms::short::Service>,
 	state_cache: Dep<rooms::state_cache::Service>,
-	typing: Dep<rooms::typing::Service>,
 }
 
 #[allow(unused, reason = "TODO refactor")]
@@ -58,33 +42,16 @@ struct SnakeSyncCache {
 
 type DbConnections<K, V> = SyncMutex<BTreeMap<K, V>>;
 type DbConnectionsKey = (OwnedUserId, OwnedDeviceId, String);
-type DbConnectionsVal = Arc<SyncMutex<SlidingSyncCache>>;
 type SnakeConnectionsKey = (OwnedUserId, OwnedDeviceId, Option<String>);
 type SnakeConnectionsVal = Arc<SyncMutex<SnakeSyncCache>>;
 
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
-			db: Data {
-				todeviceid_events: args.db["todeviceid_events"].clone(),
-				userroomid_joined: args.db["userroomid_joined"].clone(),
-				userroomid_invitestate: args.db["userroomid_invitestate"].clone(),
-				userroomid_leftstate: args.db["userroomid_leftstate"].clone(),
-				userroomid_notificationcount: args.db["userroomid_notificationcount"].clone(),
-				userroomid_highlightcount: args.db["userroomid_highlightcount"].clone(),
-				pduid_pdu: args.db["pduid_pdu"].clone(),
-				keychangeid_userid: args.db["keychangeid_userid"].clone(),
-				roomusertype_roomuserdataid: args.db["roomusertype_roomuserdataid"].clone(),
-				readreceiptid_readreceipt: args.db["readreceiptid_readreceipt"].clone(),
-				userid_lastonetimekeyupdate: args.db["userid_lastonetimekeyupdate"].clone(),
-			},
 			services: Services {
-				server: args.server.clone(),
-				short: args.depend::<rooms::short::Service>("rooms::short"),
 				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
-				typing: args.depend::<rooms::typing::Service>("rooms::typing"),
 			},
-			connections: SyncMutex::new(BTreeMap::new()),
+			wakers: Mutex::default(),
 			snake_connections: SyncMutex::new(BTreeMap::new()),
 		}))
 	}
@@ -93,6 +60,41 @@ fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
 impl Service {
+	pub async fn wait_for_wake(&self, user: &UserId) {
+		self.waker_for(user).await.notified().await;
+	}
+
+	/// Wake the target user's sync loop. Call this when something
+	/// that gets included in a legacy sync response changes.
+	///
+	/// Be careful where you call this function! In particular, don't call
+	/// it in any function that's called by `append_pdu`. `append_pdu` will call
+	/// it _after_ it's done appending a PDU, and calling it earlier can cause
+	/// hard-to-diagnose race conditions.
+	pub async fn wake(&self, user: &UserId) {
+		trace!(?user, "Waking user's sync loops");
+
+		self.waker_for(user).await.notify_waiters();
+	}
+
+	/// Wake all of our users who are joined to the specified room.
+	pub async fn wake_all_joined(&self, room: &RoomId) {
+		trace!(?room, "Waking all joined users' sync loops");
+		let mut wakers = self.wakers.lock().await;
+
+		let mut users_in_room = pin!(self.services.state_cache.active_local_users_in_room(room));
+
+		while let Some(user) = users_in_room.next().await {
+			wakers.entry(user).or_default().notify_waiters();
+		}
+	}
+
+	async fn waker_for(&self, user: &UserId) -> Arc<Notify> {
+		let mut wakers = self.wakers.lock().await;
+
+		wakers.entry(user.to_owned()).or_default().clone()
+	}
+
 	pub fn snake_connection_cached(&self, key: &SnakeConnectionsKey) -> bool {
 		self.snake_connections.lock().contains_key(key)
 	}
@@ -101,14 +103,6 @@ pub fn forget_snake_sync_connection(&self, key: &SnakeConnectionsKey) {
 		self.snake_connections.lock().remove(key);
 	}
 
-	pub fn remembered(&self, key: &DbConnectionsKey) -> bool {
-		self.connections.lock().contains_key(key)
-	}
-
-	pub fn forget_sync_request_connection(&self, key: &DbConnectionsKey) {
-		self.connections.lock().remove(key);
-	}
-
 	pub fn update_snake_sync_request_with_cache(
 		&self,
 		snake_key: &SnakeConnectionsKey,

src/service/sync/watch.rs

@@ -1,115 +0,0 @@
-use conduwuit::{Result, trace};
-use futures::{FutureExt, StreamExt, pin_mut, stream::FuturesUnordered};
-use ruma::{DeviceId, UserId};
-
-impl super::Service {
-	/// Watches for changes that might wake the sync loop for the given user +
-	/// device.
-	#[tracing::instrument(skip(self), level = "debug")]
-	pub async fn watch(&self, user_id: &UserId, device_id: &DeviceId) -> Result {
-		let userid_bytes = user_id.as_bytes().to_vec();
-		let mut userid_prefix = userid_bytes.clone();
-		userid_prefix.push(0xFF);
-
-		let mut userdeviceid_prefix = userid_prefix.clone();
-		userdeviceid_prefix.extend_from_slice(device_id.as_bytes());
-		userdeviceid_prefix.push(0xFF);
-
-		let mut futures = FuturesUnordered::new();
-
-		// Return when *any* user changed their key
-		// TODO: only send for user they share a room with
-		futures.push(self.db.todeviceid_events.watch_prefix(&userdeviceid_prefix));
-
-		futures.push(self.db.userroomid_joined.watch_prefix(&userid_prefix));
-		futures.push(self.db.userroomid_invitestate.watch_prefix(&userid_prefix));
-		futures.push(self.db.userroomid_leftstate.watch_prefix(&userid_prefix));
-		futures.push(
-			self.db
-				.userroomid_notificationcount
-				.watch_prefix(&userid_prefix),
-		);
-		futures.push(
-			self.db
-				.userroomid_highlightcount
-				.watch_prefix(&userid_prefix),
-		);
-
-		// Events for rooms we are in
-		let rooms_joined = self.services.state_cache.rooms_joined(user_id);
-
-		pin_mut!(rooms_joined);
-		while let Some(room_id) = rooms_joined.next().await {
-			let Ok(short_roomid) = self.services.short.get_shortroomid(&room_id).await else {
-				continue;
-			};
-
-			let roomid_bytes = room_id.as_bytes().to_vec();
-			let mut roomid_prefix = roomid_bytes.clone();
-			roomid_prefix.push(0xFF);
-
-			// Key changes
-			futures.push(self.db.keychangeid_userid.watch_prefix(&roomid_prefix));
-
-			// Room account data
-			let mut roomuser_prefix = roomid_prefix.clone();
-			roomuser_prefix.extend_from_slice(&userid_prefix);
-
-			futures.push(
-				self.db
-					.roomusertype_roomuserdataid
-					.watch_prefix(&roomuser_prefix),
-			);
-
-			// PDUs
-			let short_roomid = short_roomid.to_be_bytes().to_vec();
-			futures.push(self.db.pduid_pdu.watch_prefix(&short_roomid));
-
-			// EDUs
-			let typing_room_id = room_id.clone();
-			let typing_wait_for_update = async move {
-				self.services.typing.wait_for_update(&typing_room_id).await;
-			};
-
-			futures.push(typing_wait_for_update.boxed());
-			futures.push(
-				self.db
-					.readreceiptid_readreceipt
-					.watch_prefix(&roomid_prefix),
-			);
-		}
-
-		let mut globaluserdata_prefix = vec![0xFF];
-		globaluserdata_prefix.extend_from_slice(&userid_prefix);
-
-		futures.push(
-			self.db
-				.roomusertype_roomuserdataid
-				.watch_prefix(&globaluserdata_prefix),
-		);
-
-		// More key changes (used when user is not joined to any rooms)
-		futures.push(self.db.keychangeid_userid.watch_prefix(&userid_prefix));
-
-		// One time keys
-		futures.push(
-			self.db
-				.userid_lastonetimekeyupdate
-				.watch_prefix(&userid_bytes),
-		);
-
-		// Server shutdown
-		futures.push(self.services.server.until_shutdown().boxed());
-
-		if !self.services.server.running() {
-			return Ok(());
-		}
-
-		// Wait until one of them finds something
-		trace!(futures = futures.len(), "watch started");
-		futures.next().await;
-		trace!(futures = futures.len(), "watch finished");
-
-		Ok(())
-	}
-}

src/service/users/device.rs

@@ -191,6 +191,8 @@ pub async fn add_to_device_event(
 				"content": content,
 			})),
 		);
+
+		self.services.sync.wake(target_user_id).await;
 	}
 
 	/// Gets all to-device events between the two counts.
@@ -242,6 +244,8 @@ pub async fn remove_to_device_events<Until>(
 				self.db.todeviceid_events.del(key);
 			})
 			.await;
+
+		self.services.sync.wake(user_id).await;
 	}
 
 	/// Updates device metadata and increments the device list version.

src/service/users/keys.rs

@@ -56,6 +56,8 @@ pub async fn add_one_time_key(
 		let count = self.services.globals.next_count().unwrap();
 		self.db.userid_lastonetimekeyupdate.raw_put(user_id, count);
 
+		self.services.sync.wake(user_id).await;
+
 		Ok(())
 	}
 
@@ -152,6 +154,7 @@ pub async fn take_one_time_key(
 			});
 
 		if let Some(result) = one_time_key {
+			self.services.sync.wake(user_id).await;
 			return Ok(result);
 		}
 
@@ -183,6 +186,8 @@ pub async fn take_one_time_key(
 				)
 				.await?;
 			}
+
+			self.services.sync.wake(user_id).await;
 			return Ok((fallback_key_id, fallback_key_value));
 		}
 
@@ -450,9 +455,11 @@ pub async fn mark_device_key_update(&self, user_id: &UserId) {
 					None
 				}
 			})
-			.ready_for_each(|room_id| {
-				let key = (room_id, count);
+			.for_each(async |room_id| {
+				let key = (&room_id, count);
 				self.db.keychangeid_userid.put_raw(key, user_id);
+
+				self.services.sync.wake_all_joined(&room_id).await;
 			})
 			.await;
 

src/service/users/mod.rs

@@ -21,7 +21,7 @@
 use crate::{
 	Dep, account_data, admin, appservice, config, firstrun, globals, oauth, presence,
 	rooms::{self, alias, membership},
-	threepid,
+	sync, threepid,
 };
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -67,6 +67,7 @@ struct Services {
 	state: Dep<rooms::state::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
 	state_cache: Dep<rooms::state_cache::Service>,
+	sync: Dep<sync::Service>,
 	threepid: Dep<threepid::Service>,
 	timeline: Dep<rooms::timeline::Service>,
 }
@@ -119,6 +120,7 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				state_accessor: args
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
 				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
+				sync: args.depend::<sync::Service>("sync"),
 				threepid: args.depend::<threepid::Service>("threepid"),
 				timeline: args.depend::<rooms::timeline::Service>("rooms::timeline"),
 			},

src/service/users/profile.rs

@@ -4,6 +4,7 @@
 	Err, Result,
 	pdu::PartialPdu,
 	utils::{ReadyExt, stream::TryIgnore, to_canonical_object},
+	warn,
 };
 use database::{Deserialized, Ignore, Interfix, Json};
 use futures::{Stream, StreamExt};
@@ -115,18 +116,32 @@ pub async fn set_profile_field(
 
 			while let Some(room_id) = all_joined_rooms.next().await {
 				// TODO: this clobbers any custom fields on the event content
-				let mut current_membership = self
+				let mut current_membership = match self
 					.services
 					.state_accessor
 					.get_member(&room_id, user_id)
 					.await
-					.expect("should be able to fetch membership event for joined room");
-
-				assert_eq!(
-					current_membership.membership,
-					MembershipState::Join,
-					"user should be joined"
-				);
+				{
+					| Ok(current_membership)
+						if current_membership.membership == MembershipState::Join =>
+						current_membership,
+					| Ok(current_membership) => {
+						warn!(
+							?user_id,
+							?room_id,
+							"User is not joined in joined room: {current_membership:?}"
+						);
+						continue;
+					},
+					| Err(err) => {
+						warn!(
+							?user_id,
+							?room_id,
+							"Could not load membership event for joined room: {err}"
+						);
+						continue;
+					},
+				};
 
 				// If `propagate_to` is `unchanged`, and the current value of the field we're
 				// updating was changed from its global value in this room, skip it.