char/conduit
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge branch 'state-res-closure' into 'master'

Browse source 
Remove auth_cache using a closure to fetch events in state-res

See merge request famedly/conduit!108
This commit is contained in:
Timo Kösters 2021-07-02 10:56:21 +00:00
commit 5711467ad9
16 changed files with 587 additions and 487 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

89
Cargo.lock generated
View file

@ -243,6 +243,7 @@ dependencies = [
"image", "image",
"jsonwebtoken", "jsonwebtoken",
"log", "log",
"lru-cache",
"opentelemetry", "opentelemetry",
"opentelemetry-jaeger", "opentelemetry-jaeger",
"pretty_env_logger", "pretty_env_logger",
@ -271,9 +272,9 @@ dependencies = [
[[package]] [[package]]
name = "const-oid" name = "const-oid"
version = "0.5.2" version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "279bc8fc53f788a75c7804af68237d1fce02cde1e275a886a4b320604dc2aeda" checksum = "44c32f031ea41b4291d695026c023b95d59db2d8a2c7640800ed56bc8f510f22"
[[package]] [[package]]
name = "const_fn" name = "const_fn"
@ -393,9 +394,9 @@ dependencies = [
[[package]] [[package]]
name = "der" name = "der"
version = "0.3.5" version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2eeb9d92785d1facb50567852ce75d0858630630e7eabea59cf7eb7474051087" checksum = "49f215f706081a44cb702c71c39a52c05da637822e9c1645a50b7202689e982d"
dependencies = [ dependencies = [
"const-oid", "const-oid",
] ]
@ -1474,9 +1475,9 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
[[package]] [[package]]
name = "pkcs8" name = "pkcs8"
version = "0.6.1" version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c9c2f795bc591cb3384cb64082a578b89207ac92bb89c9d98c1ea2ace7cd8110" checksum = "09d156817ae0125e8aa5067710b0db24f0984830614f99875a70aa5e3b74db69"
dependencies = [ dependencies = [
"der", "der",
"spki", "spki",
@ -1882,8 +1883,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma" name = "ruma"
version = "0.1.2" version = "0.2.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"assign", "assign",
"js_int", "js_int",
@ -1903,8 +1904,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-api" name = "ruma-api"
version = "0.17.0" version = "0.17.1"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"bytes", "bytes",
"http", "http",
@ -1919,8 +1920,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-api-macros" name = "ruma-api-macros"
version = "0.17.0" version = "0.17.1"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"proc-macro-crate", "proc-macro-crate",
"proc-macro2", "proc-macro2",
@ -1930,8 +1931,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-appservice-api" name = "ruma-appservice-api"
version = "0.2.0" version = "0.3.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"ruma-api", "ruma-api",
"ruma-common", "ruma-common",
@ -1944,8 +1945,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-client-api" name = "ruma-client-api"
version = "0.10.2" version = "0.11.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"assign", "assign",
"bytes", "bytes",
@ -1964,8 +1965,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-common" name = "ruma-common"
version = "0.5.3" version = "0.5.4"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"indexmap", "indexmap",
"js_int", "js_int",
@ -1979,8 +1980,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-events" name = "ruma-events"
version = "0.22.2" version = "0.23.1"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"indoc", "indoc",
"js_int", "js_int",
@ -1994,8 +1995,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-events-macros" name = "ruma-events-macros"
version = "0.22.2" version = "0.23.1"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"proc-macro-crate", "proc-macro-crate",
"proc-macro2", "proc-macro2",
@ -2005,8 +2006,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-federation-api" name = "ruma-federation-api"
version = "0.1.0" version = "0.2.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"js_int", "js_int",
"ruma-api", "ruma-api",
@ -2020,8 +2021,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-identifiers" name = "ruma-identifiers"
version = "0.19.2" version = "0.19.4"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"paste", "paste",
"rand 0.8.3", "rand 0.8.3",
@ -2034,8 +2035,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-identifiers-macros" name = "ruma-identifiers-macros"
version = "0.19.2" version = "0.19.4"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"quote", "quote",
"ruma-identifiers-validation", "ruma-identifiers-validation",
@ -2045,12 +2046,12 @@ dependencies = [
[[package]] [[package]]
name = "ruma-identifiers-validation" name = "ruma-identifiers-validation"
version = "0.4.0" version = "0.4.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
[[package]] [[package]]
name = "ruma-identity-service-api" name = "ruma-identity-service-api"
version = "0.1.0" version = "0.2.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"js_int", "js_int",
"ruma-api", "ruma-api",
@ -2062,8 +2063,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-push-gateway-api" name = "ruma-push-gateway-api"
version = "0.1.0" version = "0.2.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"js_int", "js_int",
"ruma-api", "ruma-api",
@ -2077,8 +2078,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-serde" name = "ruma-serde"
version = "0.4.0" version = "0.4.1"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"bytes", "bytes",
"form_urlencoded", "form_urlencoded",
@ -2091,8 +2092,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-serde-macros" name = "ruma-serde-macros"
version = "0.4.0" version = "0.4.1"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"proc-macro-crate", "proc-macro-crate",
"proc-macro2", "proc-macro2",
@ -2102,8 +2103,8 @@ dependencies = [
[[package]] [[package]]
name = "ruma-signatures" name = "ruma-signatures"
version = "0.7.2" version = "0.8.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"base64 0.13.0", "base64 0.13.0",
"ed25519-dalek", "ed25519-dalek",
@ -2115,13 +2116,12 @@ dependencies = [
"sha2", "sha2",
"thiserror", "thiserror",
"tracing", "tracing",
"untrusted",
] ]
[[package]] [[package]]
name = "ruma-state-res" name = "ruma-state-res"
version = "0.1.0" version = "0.2.0"
source = "git+https://github.com/ruma/ruma?rev=5a7e2cddcf257e367465cced51442c91e8f557c9#5a7e2cddcf257e367465cced51442c91e8f557c9" source = "git+https://github.com/ruma/ruma?rev=174555857ef90d49e4b9a672be9e2fe0acdc2687#174555857ef90d49e4b9a672be9e2fe0acdc2687"
dependencies = [ dependencies = [
"itertools 0.10.0", "itertools 0.10.0",
"js_int", "js_int",
@ -2130,7 +2130,6 @@ dependencies = [
"ruma-events", "ruma-events",
"ruma-identifiers", "ruma-identifiers",
"ruma-serde", "ruma-serde",
"ruma-signatures",
"serde", "serde",
"serde_json", "serde_json",
"thiserror", "thiserror",
@ -2444,9 +2443,9 @@ checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d"
[[package]] [[package]]
name = "spki" name = "spki"
version = "0.3.0" version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9dae7e047abc519c96350e9484a96c6bf1492348af912fd3446dd2dc323f6268" checksum = "987637c5ae6b3121aba9d513f869bd2bff11c4cc086c22473befd6649c0bd521"
dependencies = [ dependencies = [
"der", "der",
] ]

4
Cargo.toml
View file

@ -8,7 +8,6 @@ repository = "https://gitlab.com/famedly/conduit"
readme = "README.md" readme = "README.md"
version = "0.1.0" version = "0.1.0"
edition = "2018" edition = "2018"
rust = "1.50"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
@ -18,7 +17,7 @@ rust = "1.50"
rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "801e04bd5369eb39e126c75f6d11e1e9597304d8", features = ["tls"] } # Used to handle requests rocket = { git = "https://github.com/SergioBenitez/Rocket.git", rev = "801e04bd5369eb39e126c75f6d11e1e9597304d8", features = ["tls"] } # Used to handle requests
# Used for matrix spec type definitions and helpers # Used for matrix spec type definitions and helpers
ruma = { git = "https://github.com/ruma/ruma", rev = "5a7e2cddcf257e367465cced51442c91e8f557c9", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] } ruma = { git = "https://github.com/ruma/ruma", rev = "174555857ef90d49e4b9a672be9e2fe0acdc2687", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] }
#ruma = { path = "../ruma/crates/ruma", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] } #ruma = { path = "../ruma/crates/ruma", features = ["compat", "rand", "appservice-api-c", "client-api", "federation-api", "push-gateway-api-c", "state-res", "unstable-pre-spec", "unstable-exhaustive-types"] }
# Used for long polling and federation sender, should be the same as rocket::tokio # Used for long polling and federation sender, should be the same as rocket::tokio
@ -73,6 +72,7 @@ tracing-subscriber = "0.2.16"
tracing-opentelemetry = "0.11.0" tracing-opentelemetry = "0.11.0"
opentelemetry-jaeger = "0.11.0" opentelemetry-jaeger = "0.11.0"
pretty_env_logger = "0.4.0" pretty_env_logger = "0.4.0"
lru-cache = "0.1.2"
[features] [features]
default = ["conduit_bin", "backend_sled"] default = ["conduit_bin", "backend_sled"]

2
rust-toolchain
View file

@ -1 +1 @@
1.50.0 1.51.0

216
src/client_server/directory.rs
View file

@ -200,84 +200,84 @@ pub async fn get_public_rooms_filtered_helper(
} }
} }
let mut all_rooms = db let mut all_rooms =
.rooms db.rooms
.public_rooms() .public_rooms()
.map(|room_id| { .map(|room_id| {
let room_id = room_id?; let room_id = room_id?;
let chunk = PublicRoomsChunk { let chunk = PublicRoomsChunk {
aliases: Vec::new(), aliases: Vec::new(),
canonical_alias: db canonical_alias: db
.rooms .rooms
.room_state_get(&room_id, &EventType::RoomCanonicalAlias, "")? .room_state_get(&room_id, &EventType::RoomCanonicalAlias, "")?
.map_or(Ok::<_, Error>(None), |s| { .map_or(Ok::<_, Error>(None), |s| {
Ok( Ok(serde_json::from_value::<
serde_json::from_value::<
Raw<canonical_alias::CanonicalAliasEventContent>, Raw<canonical_alias::CanonicalAliasEventContent>,
>(s.content) >(s.content.clone())
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()
.map_err(|_| { .map_err(|_| {
Error::bad_database("Invalid canonical alias event in database.") Error::bad_database("Invalid canonical alias event in database.")
})? })?
.alias, .alias)
) })?,
})?, name: db
name: db .rooms
.rooms .room_state_get(&room_id, &EventType::RoomName, "")?
.room_state_get(&room_id, &EventType::RoomName, "")? .map_or(Ok::<_, Error>(None), |s| {
.map_or(Ok::<_, Error>(None), |s| { Ok(serde_json::from_value::<Raw<name::NameEventContent>>(
Ok( s.content.clone(),
serde_json::from_value::<Raw<name::NameEventContent>>(s.content) )
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()
.map_err(|_| { .map_err(|_| {
Error::bad_database("Invalid room name event in database.") Error::bad_database("Invalid room name event in database.")
})? })?
.name() .name()
.map(|n| n.to_owned()), .map(|n| n.to_owned()))
) })?,
})?, num_joined_members: (db.rooms.room_members(&room_id).count() as u32).into(),
num_joined_members: (db.rooms.room_members(&room_id).count() as u32).into(), topic: db
topic: db .rooms
.rooms .room_state_get(&room_id, &EventType::RoomTopic, "")?
.room_state_get(&room_id, &EventType::RoomTopic, "")? .map_or(Ok::<_, Error>(None), |s| {
.map_or(Ok::<_, Error>(None), |s| { Ok(Some(
Ok(Some( serde_json::from_value::<Raw<topic::TopicEventContent>>(
serde_json::from_value::<Raw<topic::TopicEventContent>>(s.content) s.content.clone(),
)
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()
.map_err(|_| { .map_err(|_| {
Error::bad_database("Invalid room topic event in database.") Error::bad_database("Invalid room topic event in database.")
})? })?
.topic, .topic,
)) ))
})?, })?,
world_readable: db world_readable: db
.rooms .rooms
.room_state_get(&room_id, &EventType::RoomHistoryVisibility, "")? .room_state_get(&room_id, &EventType::RoomHistoryVisibility, "")?
.map_or(Ok::<_, Error>(false), |s| { .map_or(Ok::<_, Error>(false), |s| {
Ok(serde_json::from_value::< Ok(serde_json::from_value::<
Raw<history_visibility::HistoryVisibilityEventContent>, Raw<history_visibility::HistoryVisibilityEventContent>,
>(s.content) >(s.content.clone())
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()
.map_err(|_| { .map_err(|_| {
Error::bad_database( Error::bad_database(
"Invalid room history visibility event in database.", "Invalid room history visibility event in database.",
) )
})? })?
.history_visibility .history_visibility
== history_visibility::HistoryVisibility::WorldReadable) == history_visibility::HistoryVisibility::WorldReadable)
})?, })?,
guest_can_join: db guest_can_join: db
.rooms .rooms
.room_state_get(&room_id, &EventType::RoomGuestAccess, "")? .room_state_get(&room_id, &EventType::RoomGuestAccess, "")?
.map_or(Ok::<_, Error>(false), |s| { .map_or(Ok::<_, Error>(false), |s| {
Ok( Ok(
serde_json::from_value::<Raw<guest_access::GuestAccessEventContent>>( serde_json::from_value::<Raw<guest_access::GuestAccessEventContent>>(
s.content, s.content.clone(),
) )
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()
@ -287,61 +287,63 @@ pub async fn get_public_rooms_filtered_helper(
.guest_access .guest_access
== guest_access::GuestAccess::CanJoin, == guest_access::GuestAccess::CanJoin,
) )
})?, })?,
avatar_url: db avatar_url: db
.rooms .rooms
.room_state_get(&room_id, &EventType::RoomAvatar, "")? .room_state_get(&room_id, &EventType::RoomAvatar, "")?
.map(|s| { .map(|s| {
Ok::<_, Error>( Ok::<_, Error>(
serde_json::from_value::<Raw<avatar::AvatarEventContent>>(s.content) serde_json::from_value::<Raw<avatar::AvatarEventContent>>(
s.content.clone(),
)
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()
.map_err(|_| { .map_err(|_| {
Error::bad_database("Invalid room avatar event in database.") Error::bad_database("Invalid room avatar event in database.")
})? })?
.url, .url,
) )
}) })
.transpose()? .transpose()?
// url is now an Option<String> so we must flatten // url is now an Option<String> so we must flatten
.flatten(), .flatten(),
room_id, room_id,
}; };
Ok(chunk) Ok(chunk)
}) })
.filter_map(|r: Result<_>| r.ok()) // Filter out buggy rooms .filter_map(|r: Result<_>| r.ok()) // Filter out buggy rooms
.filter(|chunk| { .filter(|chunk| {
if let Some(query) = filter if let Some(query) = filter
.generic_search_term .generic_search_term
.as_ref() .as_ref()
.map(|q| q.to_lowercase()) .map(|q| q.to_lowercase())
{ {
if let Some(name) = &chunk.name { if let Some(name) = &chunk.name {
if name.to_lowercase().contains(&query) { if name.to_lowercase().contains(&query) {
return true; return true;
}
} }
}
if let Some(topic) = &chunk.topic { if let Some(topic) = &chunk.topic {
if topic.to_lowercase().contains(&query) { if topic.to_lowercase().contains(&query) {
return true; return true;
}
} }
}
if let Some(canonical_alias) = &chunk.canonical_alias { if let Some(canonical_alias) = &chunk.canonical_alias {
if canonical_alias.as_str().to_lowercase().contains(&query) { if canonical_alias.as_str().to_lowercase().contains(&query) {
return true; return true;
}
} }
}
false false
} else { } else {
// No search term // No search term
true true
} }
}) })
// We need to collect all, so we can sort by member count // We need to collect all, so we can sort by member count
.collect::<Vec<_>>(); .collect::<Vec<_>>();
all_rooms.sort_by(|l, r| r.num_joined_members.cmp(&l.num_joined_members)); all_rooms.sort_by(|l, r| r.num_joined_members.cmp(&l.num_joined_members));

59
src/client_server/membership.rs
View file

@ -25,7 +25,7 @@ use ruma::{
EventType, EventType,
}, },
serde::{to_canonical_value, CanonicalJsonObject, CanonicalJsonValue, Raw}, serde::{to_canonical_value, CanonicalJsonObject, CanonicalJsonValue, Raw},
state_res::{self, EventMap, RoomVersion}, state_res::{self, RoomVersion},
uint, EventId, RoomId, RoomVersionId, ServerName, UserId, uint, EventId, RoomId, RoomVersionId, ServerName, UserId,
}; };
use std::{ use std::{
@ -189,7 +189,8 @@ pub async fn kick_user_route(
ErrorKind::BadState, ErrorKind::BadState,
"Cannot kick member that's not in the room.", "Cannot kick member that's not in the room.",
))? ))?
.content, .content
.clone(),
) )
.expect("Raw::from_value always works") .expect("Raw::from_value always works")
.deserialize() .deserialize()
@ -245,11 +246,12 @@ pub async fn ban_user_route(
third_party_invite: None, third_party_invite: None,
}), }),
|event| { |event| {
let mut event = let mut event = serde_json::from_value::<Raw<member::MemberEventContent>>(
serde_json::from_value::<Raw<member::MemberEventContent>>(event.content) event.content.clone(),
.expect("Raw::from_value always works") )
.deserialize() .expect("Raw::from_value always works")
.map_err(|_| Error::bad_database("Invalid member event in database."))?; .deserialize()
.map_err(|_| Error::bad_database("Invalid member event in database."))?;
event.membership = ruma::events::room::member::MembershipState::Ban; event.membership = ruma::events::room::member::MembershipState::Ban;
Ok(event) Ok(event)
}, },
@ -295,7 +297,8 @@ pub async fn unban_user_route(
ErrorKind::BadState, ErrorKind::BadState,
"Cannot unban a user who is not banned.", "Cannot unban a user who is not banned.",
))? ))?
.content, .content
.clone(),
) )
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()
@ -753,7 +756,7 @@ pub async fn invite_helper(
let create_prev_event = if prev_events.len() == 1 let create_prev_event = if prev_events.len() == 1
&& Some(&prev_events[0]) == create_event.as_ref().map(|c| &c.event_id) && Some(&prev_events[0]) == create_event.as_ref().map(|c| &c.event_id)
{ {
create_event.map(Arc::new) create_event
} else { } else {
None None
}; };
@ -792,10 +795,10 @@ pub async fn invite_helper(
let mut unsigned = BTreeMap::new(); let mut unsigned = BTreeMap::new();
if let Some(prev_pdu) = db.rooms.room_state_get(room_id, &kind, &state_key)? { if let Some(prev_pdu) = db.rooms.room_state_get(room_id, &kind, &state_key)? {
unsigned.insert("prev_content".to_owned(), prev_pdu.content); unsigned.insert("prev_content".to_owned(), prev_pdu.content.clone());
unsigned.insert( unsigned.insert(
"prev_sender".to_owned(), "prev_sender".to_owned(),
serde_json::to_value(prev_pdu.sender).expect("UserId::to_value always works"), serde_json::to_value(&prev_pdu.sender).expect("UserId::to_value always works"),
); );
} }
@ -880,7 +883,6 @@ pub async fn invite_helper(
.await?; .await?;
let pub_key_map = RwLock::new(BTreeMap::new()); let pub_key_map = RwLock::new(BTreeMap::new());
let mut auth_cache = EventMap::new();
// We do not add the event_id field to the pdu here because of signature and hashes checks // We do not add the event_id field to the pdu here because of signature and hashes checks
let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&response.event) { let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&response.event) {
@ -903,26 +905,19 @@ pub async fn invite_helper(
) )
.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Origin field is invalid."))?; .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Origin field is invalid."))?;
let pdu_id = server_server::handle_incoming_pdu( let pdu_id =
&origin, server_server::handle_incoming_pdu(&origin, &event_id, value, true, &db, &pub_key_map)
&event_id, .await
value, .map_err(|_| {
true, Error::BadRequest(
&db, ErrorKind::InvalidParam,
&pub_key_map, "Error while handling incoming PDU.",
&mut auth_cache, )
) })?
.await .ok_or(Error::BadRequest(
.map_err(|_| { ErrorKind::InvalidParam,
Error::BadRequest( "Could not accept incoming PDU as timeline event.",
ErrorKind::InvalidParam, ))?;
"Error while handling incoming PDU.",
)
})?
.ok_or(Error::BadRequest(
ErrorKind::InvalidParam,
"Could not accept incoming PDU as timeline event.",
))?;
for server in db for server in db
.rooms .rooms

6
src/client_server/profile.rs
View file

@ -53,7 +53,8 @@ pub async fn set_displayname_route(
room.", room.",
) )
})? })?
.content, .content
.clone(),
) )
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()
@ -154,7 +155,8 @@ pub async fn set_avatar_url_route(
room.", room.",
) )
})? })?
.content, .content
.clone(),
) )
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()

6
src/client_server/room.rs
View file

@ -362,7 +362,8 @@ pub async fn upgrade_room_route(
db.rooms db.rooms
.room_state_get(&body.room_id, &EventType::RoomCreate, "")? .room_state_get(&body.room_id, &EventType::RoomCreate, "")?
.ok_or_else(|| Error::bad_database("Found room without m.room.create event."))? .ok_or_else(|| Error::bad_database("Found room without m.room.create event."))?
.content, .content
.clone(),
) )
.expect("Raw::from_value always works") .expect("Raw::from_value always works")
.deserialize() .deserialize()
@ -463,7 +464,8 @@ pub async fn upgrade_room_route(
db.rooms db.rooms
.room_state_get(&body.room_id, &EventType::RoomPowerLevels, "")? .room_state_get(&body.room_id, &EventType::RoomPowerLevels, "")?
.ok_or_else(|| Error::bad_database("Found room without m.room.create event."))? .ok_or_else(|| Error::bad_database("Found room without m.room.create event."))?
.content, .content
.clone(),
) )
.expect("database contains invalid PDU") .expect("database contains invalid PDU")
.deserialize() .deserialize()

10
src/client_server/state.rs
View file

@ -92,7 +92,7 @@ pub async fn get_state_events_route(
db.rooms db.rooms
.room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")? .room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")?
.map(|event| { .map(|event| {
serde_json::from_value::<HistoryVisibilityEventContent>(event.content) serde_json::from_value::<HistoryVisibilityEventContent>(event.content.clone())
.map_err(|_| { .map_err(|_| {
Error::bad_database( Error::bad_database(
"Invalid room history visibility event in database.", "Invalid room history visibility event in database.",
@ -139,7 +139,7 @@ pub async fn get_state_events_for_key_route(
db.rooms db.rooms
.room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")? .room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")?
.map(|event| { .map(|event| {
serde_json::from_value::<HistoryVisibilityEventContent>(event.content) serde_json::from_value::<HistoryVisibilityEventContent>(event.content.clone())
.map_err(|_| { .map_err(|_| {
Error::bad_database( Error::bad_database(
"Invalid room history visibility event in database.", "Invalid room history visibility event in database.",
@ -165,7 +165,7 @@ pub async fn get_state_events_for_key_route(
))?; ))?;
Ok(get_state_events_for_key::Response { Ok(get_state_events_for_key::Response {
content: serde_json::from_value(event.content) content: serde_json::from_value(event.content.clone())
.map_err(|_| Error::bad_database("Invalid event content in database"))?, .map_err(|_| Error::bad_database("Invalid event content in database"))?,
} }
.into()) .into())
@ -190,7 +190,7 @@ pub async fn get_state_events_for_empty_key_route(
db.rooms db.rooms
.room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")? .room_state_get(&body.room_id, &EventType::RoomHistoryVisibility, "")?
.map(|event| { .map(|event| {
serde_json::from_value::<HistoryVisibilityEventContent>(event.content) serde_json::from_value::<HistoryVisibilityEventContent>(event.content.clone())
.map_err(|_| { .map_err(|_| {
Error::bad_database( Error::bad_database(
"Invalid room history visibility event in database.", "Invalid room history visibility event in database.",
@ -216,7 +216,7 @@ pub async fn get_state_events_for_empty_key_route(
))?; ))?;
Ok(get_state_events_for_key::Response { Ok(get_state_events_for_key::Response {
content: serde_json::from_value(event.content) content: serde_json::from_value(event.content.clone())
.map_err(|_| Error::bad_database("Invalid event content in database"))?, .map_err(|_| Error::bad_database("Invalid event content in database"))?,
} }
.into()) .into())

172
src/client_server/sync.rs
View file

@ -1,21 +1,22 @@
use super::State; use super::State;
use crate::{ConduitResult, Database, Error, Result, Ruma}; use crate::{ConduitResult, Database, Error, Result, Ruma, RumaResponse};
use log::error; use log::error;
use ruma::{ use ruma::{
api::client::r0::sync::sync_events, api::client::r0::{sync::sync_events, uiaa::UiaaResponse},
events::{room::member::MembershipState, AnySyncEphemeralRoomEvent, EventType}, events::{room::member::MembershipState, AnySyncEphemeralRoomEvent, EventType},
serde::Raw, serde::Raw,
RoomId, UserId, DeviceId, RoomId, UserId,
}; };
#[cfg(feature = "conduit_bin")]
use rocket::{get, tokio};
use std::{ use std::{
collections::{hash_map, BTreeMap, HashMap, HashSet}, collections::{btree_map::Entry, hash_map, BTreeMap, HashMap, HashSet},
convert::{TryFrom, TryInto}, convert::{TryFrom, TryInto},
sync::Arc, sync::Arc,
time::Duration, time::Duration,
}; };
use tokio::sync::watch::Sender;
#[cfg(feature = "conduit_bin")]
use rocket::{get, tokio};
/// # `GET /_matrix/client/r0/sync` /// # `GET /_matrix/client/r0/sync`
/// ///
@ -36,21 +37,134 @@ use std::{
pub async fn sync_events_route( pub async fn sync_events_route(
db: State<'_, Arc<Database>>, db: State<'_, Arc<Database>>,
body: Ruma<sync_events::Request<'_>>, body: Ruma<sync_events::Request<'_>>,
) -> ConduitResult<sync_events::Response> { ) -> std::result::Result<RumaResponse<sync_events::Response>, RumaResponse<UiaaResponse>> {
let sender_user = body.sender_user.as_ref().expect("user is authenticated"); let sender_user = body.sender_user.as_ref().expect("user is authenticated");
let sender_device = body.sender_device.as_ref().expect("user is authenticated"); let sender_device = body.sender_device.as_ref().expect("user is authenticated");
let mut rx = match db
.globals
.sync_receivers
.write()
.unwrap()
.entry((sender_user.clone(), sender_device.clone()))
{
Entry::Vacant(v) => {
let (tx, rx) = tokio::sync::watch::channel(None);
tokio::spawn(sync_helper_wrapper(
Arc::clone(&db),
sender_user.clone(),
sender_device.clone(),
body.since.clone(),
body.full_state,
body.timeout,
tx,
));
v.insert((body.since.clone(), rx)).1.clone()
}
Entry::Occupied(mut o) => {
if o.get().0 != body.since {
let (tx, rx) = tokio::sync::watch::channel(None);
tokio::spawn(sync_helper_wrapper(
Arc::clone(&db),
sender_user.clone(),
sender_device.clone(),
body.since.clone(),
body.full_state,
body.timeout,
tx,
));
o.insert((body.since.clone(), rx.clone()));
rx
} else {
o.get().1.clone()
}
}
};
let we_have_to_wait = rx.borrow().is_none();
if we_have_to_wait {
let _ = rx.changed().await;
}
let result = match rx
.borrow()
.as_ref()
.expect("When sync channel changes it's always set to some")
{
Ok(response) => Ok(response.clone()),
Err(error) => Err(error.to_response()),
};
result
}
pub async fn sync_helper_wrapper(
db: Arc<Database>,
sender_user: UserId,
sender_device: Box<DeviceId>,
since: Option<String>,
full_state: bool,
timeout: Option<Duration>,
tx: Sender<Option<ConduitResult<sync_events::Response>>>,
) {
let r = sync_helper(
Arc::clone(&db),
sender_user.clone(),
sender_device.clone(),
since.clone(),
full_state,
timeout,
)
.await;
if let Ok((_, caching_allowed)) = r {
if !caching_allowed {
match db
.globals
.sync_receivers
.write()
.unwrap()
.entry((sender_user, sender_device))
{
Entry::Occupied(o) => {
// Only remove if the device didn't start a different /sync already
if o.get().0 == since {
o.remove();
}
}
Entry::Vacant(_) => {}
}
}
}
let _ = tx.send(Some(r.map(|(r, _)| r.into())));
}
async fn sync_helper(
db: Arc<Database>,
sender_user: UserId,
sender_device: Box<DeviceId>,
since: Option<String>,
full_state: bool,
timeout: Option<Duration>,
// bool = caching allowed
) -> std::result::Result<(sync_events::Response, bool), Error> {
// TODO: match body.set_presence { // TODO: match body.set_presence {
db.rooms.edus.ping_presence(&sender_user)?; db.rooms.edus.ping_presence(&sender_user)?;
// Setup watchers, so if there's no response, we can wait for them // Setup watchers, so if there's no response, we can wait for them
let watcher = db.watch(sender_user, sender_device); let watcher = db.watch(&sender_user, &sender_device);
let next_batch = db.globals.current_count()?.to_string(); let next_batch = db.globals.current_count()?;
let next_batch_string = next_batch.to_string();
let mut joined_rooms = BTreeMap::new(); let mut joined_rooms = BTreeMap::new();
let since = body let since = since
.since
.clone() .clone()
.and_then(|string| string.parse().ok()) .and_then(|string| string.parse().ok())
.unwrap_or(0); .unwrap_or(0);
@ -114,10 +228,11 @@ pub async fn sync_events_route(
// since and the current room state, meaning there should be no updates. // since and the current room state, meaning there should be no updates.
// The inner Option is None when there is an event, but there is no state hash associated // The inner Option is None when there is an event, but there is no state hash associated
// with it. This can happen for the RoomCreate event, so all updates should arrive. // with it. This can happen for the RoomCreate event, so all updates should arrive.
let first_pdu_before_since = db.rooms.pdus_until(sender_user, &room_id, since).next(); let first_pdu_before_since = db.rooms.pdus_until(&sender_user, &room_id, since).next();
let pdus_after_since = db let pdus_after_since = db
.rooms .rooms
.pdus_after(sender_user, &room_id, since) .pdus_after(&sender_user, &room_id, since)
.next() .next()
.is_some(); .is_some();
@ -256,11 +371,11 @@ pub async fn sync_events_route(
.flatten() .flatten()
.filter(|user_id| { .filter(|user_id| {
// Don't send key updates from the sender to the sender // Don't send key updates from the sender to the sender
sender_user != user_id &sender_user != user_id
}) })
.filter(|user_id| { .filter(|user_id| {
// Only send keys if the sender doesn't share an encrypted room with the target already // Only send keys if the sender doesn't share an encrypted room with the target already
!share_encrypted_room(&db, sender_user, user_id, &room_id) !share_encrypted_room(&db, &sender_user, user_id, &room_id)
.unwrap_or(false) .unwrap_or(false)
}), }),
); );
@ -335,7 +450,7 @@ pub async fn sync_events_route(
let state_events = if joined_since_last_sync { let state_events = if joined_since_last_sync {
current_state current_state
.into_iter() .iter()
.map(|(_, pdu)| pdu.to_sync_state_event()) .map(|(_, pdu)| pdu.to_sync_state_event())
.collect() .collect()
} else { } else {
@ -520,7 +635,7 @@ pub async fn sync_events_route(
account_data: sync_events::RoomAccountData { events: Vec::new() }, account_data: sync_events::RoomAccountData { events: Vec::new() },
timeline: sync_events::Timeline { timeline: sync_events::Timeline {
limited: false, limited: false,
prev_batch: Some(next_batch.clone()), prev_batch: Some(next_batch_string.clone()),
events: Vec::new(), events: Vec::new(),
}, },
state: sync_events::State { state: sync_events::State {
@ -573,10 +688,10 @@ pub async fn sync_events_route(
// Remove all to-device events the device received *last time* // Remove all to-device events the device received *last time*
db.users db.users
.remove_to_device_events(sender_user, sender_device, since)?; .remove_to_device_events(&sender_user, &sender_device, since)?;
let response = sync_events::Response { let response = sync_events::Response {
next_batch, next_batch: next_batch_string,
rooms: sync_events::Rooms { rooms: sync_events::Rooms {
leave: left_rooms, leave: left_rooms,
join: joined_rooms, join: joined_rooms,
@ -604,20 +719,22 @@ pub async fn sync_events_route(
changed: device_list_updates.into_iter().collect(), changed: device_list_updates.into_iter().collect(),
left: device_list_left.into_iter().collect(), left: device_list_left.into_iter().collect(),
}, },
device_one_time_keys_count: if db.users.last_one_time_keys_update(sender_user)? > since device_one_time_keys_count: if db.users.last_one_time_keys_update(&sender_user)? > since
|| since == 0 || since == 0
{ {
db.users.count_one_time_keys(sender_user, sender_device)? db.users.count_one_time_keys(&sender_user, &sender_device)?
} else { } else {
BTreeMap::new() BTreeMap::new()
}, },
to_device: sync_events::ToDevice { to_device: sync_events::ToDevice {
events: db.users.get_to_device_events(sender_user, sender_device)?, events: db
.users
.get_to_device_events(&sender_user, &sender_device)?,
}, },
}; };
// TODO: Retry the endpoint instead of returning (waiting for #118) // TODO: Retry the endpoint instead of returning (waiting for #118)
if !body.full_state if !full_state
&& response.rooms.is_empty() && response.rooms.is_empty()
&& response.presence.is_empty() && response.presence.is_empty()
&& response.account_data.is_empty() && response.account_data.is_empty()
@ -627,14 +744,15 @@ pub async fn sync_events_route(
{ {
// Hang a few seconds so requests are not spammed // Hang a few seconds so requests are not spammed
// Stop hanging if new info arrives // Stop hanging if new info arrives
let mut duration = body.timeout.unwrap_or_default(); let mut duration = timeout.unwrap_or_default();
if duration.as_secs() > 30 { if duration.as_secs() > 30 {
duration = Duration::from_secs(30); duration = Duration::from_secs(30);
} }
let _ = tokio::time::timeout(duration, watcher).await; let _ = tokio::time::timeout(duration, watcher).await;
Ok((response, false))
} else {
Ok((response, since != next_batch)) // Only cache if we made progress
} }
Ok(response.into())
} }
#[tracing::instrument(skip(db))] #[tracing::instrument(skip(db))]

2
src/database.rs
View file

@ -18,6 +18,7 @@ use crate::{utils, Error, Result};
use abstraction::DatabaseEngine; use abstraction::DatabaseEngine;
use directories::ProjectDirs; use directories::ProjectDirs;
use log::error; use log::error;
use lru_cache::LruCache;
use rocket::futures::{channel::mpsc, stream::FuturesUnordered, StreamExt}; use rocket::futures::{channel::mpsc, stream::FuturesUnordered, StreamExt};
use ruma::{DeviceId, ServerName, UserId}; use ruma::{DeviceId, ServerName, UserId};
use serde::Deserialize; use serde::Deserialize;
@ -194,6 +195,7 @@ impl Database {
eventid_outlierpdu: builder.open_tree("eventid_outlierpdu")?, eventid_outlierpdu: builder.open_tree("eventid_outlierpdu")?,
prevevent_parent: builder.open_tree("prevevent_parent")?, prevevent_parent: builder.open_tree("prevevent_parent")?,
pdu_cache: RwLock::new(LruCache::new(1_000_000)),
}, },
account_data: account_data::AccountData { account_data: account_data::AccountData {
roomuserdataid_accountdata: builder.open_tree("roomuserdataid_accountdata")?, roomuserdataid_accountdata: builder.open_tree("roomuserdataid_accountdata")?,

19
src/database/globals.rs
View file

@ -1,8 +1,11 @@
use crate::{database::Config, utils, Error, Result}; use crate::{database::Config, utils, ConduitResult, Error, Result};
use log::{error, info}; use log::{error, info};
use ruma::{ use ruma::{
api::federation::discovery::{ServerSigningKeys, VerifyKey}, api::{
EventId, MilliSecondsSinceUnixEpoch, ServerName, ServerSigningKeyId, client::r0::sync::sync_events,
federation::discovery::{ServerSigningKeys, VerifyKey},
},
DeviceId, EventId, MilliSecondsSinceUnixEpoch, ServerName, ServerSigningKeyId, UserId,
}; };
use rustls::{ServerCertVerifier, WebPKIVerifier}; use rustls::{ServerCertVerifier, WebPKIVerifier};
use std::{ use std::{
@ -35,6 +38,15 @@ pub struct Globals {
pub bad_event_ratelimiter: Arc<RwLock<BTreeMap<EventId, RateLimitState>>>, pub bad_event_ratelimiter: Arc<RwLock<BTreeMap<EventId, RateLimitState>>>,
pub bad_signature_ratelimiter: Arc<RwLock<BTreeMap<Vec<String>, RateLimitState>>>, pub bad_signature_ratelimiter: Arc<RwLock<BTreeMap<Vec<String>, RateLimitState>>>,
pub servername_ratelimiter: Arc<RwLock<BTreeMap<Box<ServerName>, Arc<Semaphore>>>>, pub servername_ratelimiter: Arc<RwLock<BTreeMap<Box<ServerName>, Arc<Semaphore>>>>,
pub sync_receivers: RwLock<
BTreeMap<
(UserId, Box<DeviceId>),
(
Option<String>,
tokio::sync::watch::Receiver<Option<ConduitResult<sync_events::Response>>>,
), // since, rx
>,
>,
} }
struct MatrixServerVerifier { struct MatrixServerVerifier {
@ -155,6 +167,7 @@ impl Globals {
bad_event_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())), bad_event_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())),
bad_signature_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())), bad_signature_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())),
servername_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())), servername_ratelimiter: Arc::new(RwLock::new(BTreeMap::new())),
sync_receivers: RwLock::new(BTreeMap::new()),
}; };
fs::create_dir_all(s.get_media_folder())?; fs::create_dir_all(s.get_media_folder())?;

2
src/database/pusher.rs
View file

@ -203,7 +203,7 @@ pub fn get_actions<'a>(
.rooms .rooms
.room_state_get(&pdu.room_id, &EventType::RoomPowerLevels, "")? .room_state_get(&pdu.room_id, &EventType::RoomPowerLevels, "")?
.map(|ev| { .map(|ev| {
serde_json::from_value(ev.content) serde_json::from_value(ev.content.clone())
.map_err(|_| Error::bad_database("invalid m.room.power_levels event")) .map_err(|_| Error::bad_database("invalid m.room.power_levels event"))
}) })
.transpose()? .transpose()?

69
src/database/rooms.rs
View file

@ -5,6 +5,7 @@ use member::MembershipState;
use crate::{pdu::PduBuilder, utils, Database, Error, PduEvent, Result}; use crate::{pdu::PduBuilder, utils, Database, Error, PduEvent, Result};
use log::{debug, error, warn}; use log::{debug, error, warn};
use lru_cache::LruCache;
use regex::Regex; use regex::Regex;
use ring::digest; use ring::digest;
use ruma::{ use ruma::{
@ -23,7 +24,7 @@ use std::{
collections::{BTreeMap, HashMap, HashSet}, collections::{BTreeMap, HashMap, HashSet},
convert::{TryFrom, TryInto}, convert::{TryFrom, TryInto},
mem, mem,
sync::Arc, sync::{Arc, RwLock},
}; };
use super::{abstraction::Tree, admin::AdminCommand, pusher}; use super::{abstraction::Tree, admin::AdminCommand, pusher};
@ -81,6 +82,8 @@ pub struct Rooms {
/// RoomId + EventId -> Parent PDU EventId. /// RoomId + EventId -> Parent PDU EventId.
pub(super) prevevent_parent: Arc<dyn Tree>, pub(super) prevevent_parent: Arc<dyn Tree>,
pub(super) pdu_cache: RwLock<LruCache<EventId, Arc<PduEvent>>>,
} }
impl Rooms { impl Rooms {
@ -105,8 +108,8 @@ impl Rooms {
pub fn state_full( pub fn state_full(
&self, &self,
shortstatehash: u64, shortstatehash: u64,
) -> Result<BTreeMap<(EventType, String), PduEvent>> { ) -> Result<BTreeMap<(EventType, String), Arc<PduEvent>>> {
Ok(self let state = self
.stateid_shorteventid .stateid_shorteventid
.scan_prefix(shortstatehash.to_be_bytes().to_vec()) .scan_prefix(shortstatehash.to_be_bytes().to_vec())
.map(|(_, bytes)| self.shorteventid_eventid.get(&bytes).ok().flatten()) .map(|(_, bytes)| self.shorteventid_eventid.get(&bytes).ok().flatten())
@ -133,7 +136,9 @@ impl Rooms {
)) ))
}) })
.filter_map(|r| r.ok()) .filter_map(|r| r.ok())
.collect()) .collect();
Ok(state)
} }
/// Returns a single PDU from `room_id` with key (`event_type`, `state_key`). /// Returns a single PDU from `room_id` with key (`event_type`, `state_key`).
@ -179,7 +184,7 @@ impl Rooms {
shortstatehash: u64, shortstatehash: u64,
event_type: &EventType, event_type: &EventType,
state_key: &str, state_key: &str,
) -> Result<Option<PduEvent>> { ) -> Result<Option<Arc<PduEvent>>> {
self.state_get_id(shortstatehash, event_type, state_key)? self.state_get_id(shortstatehash, event_type, state_key)?
.map_or(Ok(None), |event_id| self.get_pdu(&event_id)) .map_or(Ok(None), |event_id| self.get_pdu(&event_id))
} }
@ -234,7 +239,7 @@ impl Rooms {
let mut events = StateMap::new(); let mut events = StateMap::new();
for (event_type, state_key) in auth_events { for (event_type, state_key) in auth_events {
if let Some(pdu) = self.room_state_get(room_id, &event_type, &state_key)? { if let Some(pdu) = self.room_state_get(room_id, &event_type, &state_key)? {
events.insert((event_type, state_key), Arc::new(pdu)); events.insert((event_type, state_key), pdu);
} else { } else {
// This is okay because when creating a new room some events were not created yet // This is okay because when creating a new room some events were not created yet
debug!( debug!(
@ -396,7 +401,7 @@ impl Rooms {
pub fn room_state_full( pub fn room_state_full(
&self, &self,
room_id: &RoomId, room_id: &RoomId,
) -> Result<BTreeMap<(EventType, String), PduEvent>> { ) -> Result<BTreeMap<(EventType, String), Arc<PduEvent>>> {
if let Some(current_shortstatehash) = self.current_shortstatehash(room_id)? { if let Some(current_shortstatehash) = self.current_shortstatehash(room_id)? {
self.state_full(current_shortstatehash) self.state_full(current_shortstatehash)
} else { } else {
@ -426,7 +431,7 @@ impl Rooms {
room_id: &RoomId, room_id: &RoomId,
event_type: &EventType, event_type: &EventType,
state_key: &str, state_key: &str,
) -> Result<Option<PduEvent>> { ) -> Result<Option<Arc<PduEvent>>> {
if let Some(current_shortstatehash) = self.current_shortstatehash(room_id)? { if let Some(current_shortstatehash) = self.current_shortstatehash(room_id)? {
self.state_get(current_shortstatehash, event_type, state_key) self.state_get(current_shortstatehash, event_type, state_key)
} else { } else {
@ -514,21 +519,42 @@ impl Rooms {
/// Returns the pdu. /// Returns the pdu.
/// ///
/// Checks the `eventid_outlierpdu` Tree if not found in the timeline. /// Checks the `eventid_outlierpdu` Tree if not found in the timeline.
pub fn get_pdu(&self, event_id: &EventId) -> Result<Option<PduEvent>> { pub fn get_pdu(&self, event_id: &EventId) -> Result<Option<Arc<PduEvent>>> {
self.eventid_pduid if let Some(p) = self.pdu_cache.write().unwrap().get_mut(&event_id) {
return Ok(Some(Arc::clone(p)));
}
if let Some(pdu) = self
.eventid_pduid
.get(event_id.as_bytes())? .get(event_id.as_bytes())?
.map_or_else::<Result<_>, _, _>( .map_or_else::<Result<_>, _, _>(
|| self.eventid_outlierpdu.get(event_id.as_bytes()), || {
let r = self.eventid_outlierpdu.get(event_id.as_bytes());
r
},
|pduid| { |pduid| {
Ok(Some(self.pduid_pdu.get(&pduid)?.ok_or_else(|| { let r = Ok(Some(self.pduid_pdu.get(&pduid)?.ok_or_else(|| {
Error::bad_database("Invalid pduid in eventid_pduid.") Error::bad_database("Invalid pduid in eventid_pduid.")
})?)) })?));
r
}, },
)? )?
.map(|pdu| { .map(|pdu| {
serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db.")) let r = serde_json::from_slice(&pdu)
.map_err(|_| Error::bad_database("Invalid PDU in db."))
.map(Arc::new);
r
}) })
.transpose() .transpose()?
{
self.pdu_cache
.write()
.unwrap()
.insert(event_id.clone(), Arc::clone(&pdu));
Ok(Some(pdu))
} else {
Ok(None)
}
} }
/// Returns the pdu. /// Returns the pdu.
@ -663,7 +689,7 @@ impl Rooms {
unsigned.insert( unsigned.insert(
"prev_content".to_owned(), "prev_content".to_owned(),
CanonicalJsonValue::Object( CanonicalJsonValue::Object(
utils::to_canonical_object(prev_state.content) utils::to_canonical_object(prev_state.content.clone())
.expect("event is valid, we just created it"), .expect("event is valid, we just created it"),
), ),
); );
@ -1204,7 +1230,7 @@ impl Rooms {
let create_prev_event = if prev_events.len() == 1 let create_prev_event = if prev_events.len() == 1
&& Some(&prev_events[0]) == create_event.as_ref().map(|c| &c.event_id) && Some(&prev_events[0]) == create_event.as_ref().map(|c| &c.event_id)
{ {
create_event.map(Arc::new) create_event
} else { } else {
None None
}; };
@ -1235,10 +1261,10 @@ impl Rooms {
let mut unsigned = unsigned.unwrap_or_default(); let mut unsigned = unsigned.unwrap_or_default();
if let Some(state_key) = &state_key { if let Some(state_key) = &state_key {
if let Some(prev_pdu) = self.room_state_get(&room_id, &event_type, &state_key)? { if let Some(prev_pdu) = self.room_state_get(&room_id, &event_type, &state_key)? {
unsigned.insert("prev_content".to_owned(), prev_pdu.content); unsigned.insert("prev_content".to_owned(), prev_pdu.content.clone());
unsigned.insert( unsigned.insert(
"prev_sender".to_owned(), "prev_sender".to_owned(),
serde_json::to_value(prev_pdu.sender).expect("UserId::to_value always works"), serde_json::to_value(&prev_pdu.sender).expect("UserId::to_value always works"),
); );
} }
} }
@ -1583,7 +1609,7 @@ impl Rooms {
.and_then(|create| { .and_then(|create| {
serde_json::from_value::< serde_json::from_value::<
Raw<ruma::events::room::create::CreateEventContent>, Raw<ruma::events::room::create::CreateEventContent>,
>(create.content) >(create.content.clone())
.expect("Raw::from_value always works") .expect("Raw::from_value always works")
.deserialize() .deserialize()
.ok() .ok()
@ -1764,7 +1790,8 @@ impl Rooms {
ErrorKind::BadState, ErrorKind::BadState,
"Cannot leave a room you are not a member of.", "Cannot leave a room you are not a member of.",
))? ))?
.content, .content
.clone(),
) )
.expect("from_value::<Raw<..>> can never fail") .expect("from_value::<Raw<..>> can never fail")
.deserialize() .deserialize()

31
src/error.rs
View file

@ -61,7 +61,6 @@ pub enum Error {
BadDatabase(&'static str), BadDatabase(&'static str),
#[error("uiaa")] #[error("uiaa")]
Uiaa(UiaaInfo), Uiaa(UiaaInfo),
#[error("{0}: {1}")] #[error("{0}: {1}")]
BadRequest(ErrorKind, &'static str), BadRequest(ErrorKind, &'static str),
#[error("{0}")] #[error("{0}")]
@ -80,19 +79,16 @@ impl Error {
} }
} }
#[cfg(feature = "conduit_bin")] impl Error {
impl<'r, 'o> Responder<'r, 'o> for Error pub fn to_response(&self) -> RumaResponse<UiaaResponse> {
where
'o: 'r,
{
fn respond_to(self, r: &'r Request<'_>) -> response::Result<'o> {
if let Self::Uiaa(uiaainfo) = self { if let Self::Uiaa(uiaainfo) = self {
return RumaResponse::from(UiaaResponse::AuthResponse(uiaainfo)).respond_to(r); return RumaResponse(UiaaResponse::AuthResponse(uiaainfo.clone()));
} }
if let Self::FederationError(origin, mut error) = self { if let Self::FederationError(origin, error) = self {
let mut error = error.clone();
error.message = format!("Answer from {}: {}", origin, error.message); error.message = format!("Answer from {}: {}", origin, error.message);
return RumaResponse::from(error).respond_to(r); return RumaResponse(UiaaResponse::MatrixError(error));
} }
let message = format!("{}", self); let message = format!("{}", self);
@ -119,11 +115,20 @@ where
warn!("{}: {}", status_code, message); warn!("{}: {}", status_code, message);
RumaResponse::from(RumaError { RumaResponse(UiaaResponse::MatrixError(RumaError {
kind, kind,
message, message,
status_code, status_code,
}) }))
.respond_to(r) }
}
#[cfg(feature = "conduit_bin")]
impl<'r, 'o> Responder<'r, 'o> for Error
where
'o: 'r,
{
fn respond_to(self, r: &'r Request<'_>) -> response::Result<'o> {
self.to_response().respond_to(r)
} }
} }

77
src/ruma_wrapper.rs
View file

@ -1,6 +1,6 @@
use crate::Error; use crate::Error;
use ruma::{ use ruma::{
api::OutgoingResponse, api::{client::r0::uiaa::UiaaResponse, OutgoingResponse},
identifiers::{DeviceId, UserId}, identifiers::{DeviceId, UserId},
signatures::CanonicalJsonValue, signatures::CanonicalJsonValue,
Outgoing, ServerName, Outgoing, ServerName,
@ -335,49 +335,60 @@ impl<T: Outgoing> Deref for Ruma<T> {
/// This struct converts ruma responses into rocket http responses. /// This struct converts ruma responses into rocket http responses.
pub type ConduitResult<T> = std::result::Result<RumaResponse<T>, Error>; pub type ConduitResult<T> = std::result::Result<RumaResponse<T>, Error>;
pub struct RumaResponse<T: OutgoingResponse>(pub T); pub fn response<T: OutgoingResponse>(response: RumaResponse<T>) -> response::Result<'static> {
let http_response = response
.0
.try_into_http_response::<Vec<u8>>()
.map_err(|_| Status::InternalServerError)?;
impl<T: OutgoingResponse> From<T> for RumaResponse<T> { let mut response = rocket::response::Response::build();
let status = http_response.status();
response.raw_status(status.into(), "");
for header in http_response.headers() {
response.raw_header(header.0.to_string(), header.1.to_str().unwrap().to_owned());
}
let http_body = http_response.into_body();
response.sized_body(http_body.len(), Cursor::new(http_body));
response.raw_header("Access-Control-Allow-Origin", "*");
response.raw_header(
"Access-Control-Allow-Methods",
"GET, POST, PUT, DELETE, OPTIONS",
);
response.raw_header(
"Access-Control-Allow-Headers",
"Origin, X-Requested-With, Content-Type, Accept, Authorization",
);
response.raw_header("Access-Control-Max-Age", "86400");
response.ok()
}
#[derive(Clone)]
pub struct RumaResponse<T>(pub T);
impl<T> From<T> for RumaResponse<T> {
fn from(t: T) -> Self { fn from(t: T) -> Self {
Self(t) Self(t)
} }
} }
impl From<Error> for RumaResponse<UiaaResponse> {
fn from(t: Error) -> Self {
t.to_response()
}
}
#[cfg(feature = "conduit_bin")] #[cfg(feature = "conduit_bin")]
impl<'r, 'o, T> Responder<'r, 'o> for RumaResponse<T> impl<'r, 'o, T> Responder<'r, 'o> for RumaResponse<T>
where where
T: Send + OutgoingResponse,
'o: 'r, 'o: 'r,
T: OutgoingResponse,
{ {
fn respond_to(self, _: &'r Request<'_>) -> response::Result<'o> { fn respond_to(self, _: &'r Request<'_>) -> response::Result<'o> {
let http_response = self response(self)
.0
.try_into_http_response::<Vec<u8>>()
.map_err(|_| Status::InternalServerError)?;
let mut response = rocket::response::Response::build();
let status = http_response.status();
response.raw_status(status.into(), "");
for header in http_response.headers() {
response.raw_header(header.0.to_string(), header.1.to_str().unwrap().to_owned());
}
let http_body = http_response.into_body();
response.sized_body(http_body.len(), Cursor::new(http_body));
response.raw_header("Access-Control-Allow-Origin", "*");
response.raw_header(
"Access-Control-Allow-Methods",
"GET, POST, PUT, DELETE, OPTIONS",
);
response.raw_header(
"Access-Control-Allow-Headers",
"Origin, X-Requested-With, Content-Type, Accept, Authorization",
);
response.raw_header("Access-Control-Max-Age", "86400");
response.ok()
} }
} }

310
src/server_server.rs
View file

@ -45,7 +45,7 @@ use ruma::{
receipt::ReceiptType, receipt::ReceiptType,
serde::Raw, serde::Raw,
signatures::{CanonicalJsonObject, CanonicalJsonValue}, signatures::{CanonicalJsonObject, CanonicalJsonValue},
state_res::{self, Event, EventMap, RoomVersion, StateMap}, state_res::{self, Event, RoomVersion, StateMap},
to_device::DeviceIdOrAllDevices, to_device::DeviceIdOrAllDevices,
uint, EventId, MilliSecondsSinceUnixEpoch, RoomId, RoomVersionId, ServerName, uint, EventId, MilliSecondsSinceUnixEpoch, RoomId, RoomVersionId, ServerName,
ServerSigningKeyId, UserId, ServerSigningKeyId, UserId,
@ -612,7 +612,7 @@ pub async fn send_transaction_message_route(
// TODO: This could potentially also be some sort of trie (suffix tree) like structure so // TODO: This could potentially also be some sort of trie (suffix tree) like structure so
// that once an auth event is known it would know (using indexes maybe) all of the auth // that once an auth event is known it would know (using indexes maybe) all of the auth
// events that it references. // events that it references.
let mut auth_cache = EventMap::new(); // let mut auth_cache = EventMap::new();
for pdu in &body.pdus { for pdu in &body.pdus {
// We do not add the event_id field to the pdu here because of signature and hashes checks // We do not add the event_id field to the pdu here because of signature and hashes checks
@ -627,17 +627,9 @@ pub async fn send_transaction_message_route(
let start_time = Instant::now(); let start_time = Instant::now();
resolved_map.insert( resolved_map.insert(
event_id.clone(), event_id.clone(),
handle_incoming_pdu( handle_incoming_pdu(&body.origin, &event_id, value, true, &db, &pub_key_map)
&body.origin, .await
&event_id, .map(|_| ()),
value,
true,
&db,
&pub_key_map,
&mut auth_cache,
)
.await
.map(|_| ()),
); );
let elapsed = start_time.elapsed(); let elapsed = start_time.elapsed();
@ -820,7 +812,6 @@ pub fn handle_incoming_pdu<'a>(
is_timeline_event: bool, is_timeline_event: bool,
db: &'a Database, db: &'a Database,
pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, String>>>, pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, String>>>,
auth_cache: &'a mut EventMap<Arc<PduEvent>>,
) -> AsyncRecursiveResult<'a, Option<Vec<u8>>, String> { ) -> AsyncRecursiveResult<'a, Option<Vec<u8>>, String> {
Box::pin(async move { Box::pin(async move {
// TODO: For RoomVersion6 we must check that Raw<..> is canonical do we anywhere?: https://matrix.org/docs/spec/rooms/v6#canonical-json // TODO: For RoomVersion6 we must check that Raw<..> is canonical do we anywhere?: https://matrix.org/docs/spec/rooms/v6#canonical-json
@ -908,15 +899,9 @@ pub fn handle_incoming_pdu<'a>(
// 5. Reject "due to auth events" if can't get all the auth events or some of the auth events are also rejected "due to auth events" // 5. Reject "due to auth events" if can't get all the auth events or some of the auth events are also rejected "due to auth events"
// EDIT: Step 5 is not applied anymore because it failed too often // EDIT: Step 5 is not applied anymore because it failed too often
debug!("Fetching auth events for {}", incoming_pdu.event_id); debug!("Fetching auth events for {}", incoming_pdu.event_id);
fetch_and_handle_events( fetch_and_handle_events(db, origin, &incoming_pdu.auth_events, pub_key_map)
db, .await
origin, .map_err(|e| e.to_string())?;
&incoming_pdu.auth_events,
pub_key_map,
auth_cache,
)
.await
.map_err(|e| e.to_string())?;
// 6. Reject "due to auth events" if the event doesn't pass auth based on the auth events // 6. Reject "due to auth events" if the event doesn't pass auth based on the auth events
debug!( debug!(
@ -927,9 +912,13 @@ pub fn handle_incoming_pdu<'a>(
// Build map of auth events // Build map of auth events
let mut auth_events = BTreeMap::new(); let mut auth_events = BTreeMap::new();
for id in &incoming_pdu.auth_events { for id in &incoming_pdu.auth_events {
let auth_event = auth_cache.get(id).ok_or_else(|| { let auth_event = db
"Auth event not found, event failed recursive auth checks.".to_string() .rooms
})?; .get_pdu(id)
.map_err(|e| e.to_string())?
.ok_or_else(|| {
"Auth event not found, event failed recursive auth checks.".to_string()
})?;
match auth_events.entry(( match auth_events.entry((
auth_event.kind.clone(), auth_event.kind.clone(),
@ -963,10 +952,10 @@ pub fn handle_incoming_pdu<'a>(
let previous_create = if incoming_pdu.auth_events.len() == 1 let previous_create = if incoming_pdu.auth_events.len() == 1
&& incoming_pdu.prev_events == incoming_pdu.auth_events && incoming_pdu.prev_events == incoming_pdu.auth_events
{ {
auth_cache db.rooms
.get(&incoming_pdu.auth_events[0]) .get_pdu(&incoming_pdu.auth_events[0])
.cloned() .map_err(|e| e.to_string())?
.filter(|maybe_create| **maybe_create == create_event) .filter(|maybe_create| **maybe_create == *create_event)
} else { } else {
None None
}; };
@ -1008,7 +997,6 @@ pub fn handle_incoming_pdu<'a>(
debug!("Requesting state at event."); debug!("Requesting state at event.");
let mut state_at_incoming_event = None; let mut state_at_incoming_event = None;
let mut incoming_auth_events = Vec::new();
if incoming_pdu.prev_events.len() == 1 { if incoming_pdu.prev_events.len() == 1 {
let prev_event = &incoming_pdu.prev_events[0]; let prev_event = &incoming_pdu.prev_events[0];
@ -1031,7 +1019,7 @@ pub fn handle_incoming_pdu<'a>(
state_vec.push(prev_event.clone()); state_vec.push(prev_event.clone());
} }
state_at_incoming_event = Some( state_at_incoming_event = Some(
fetch_and_handle_events(db, origin, &state_vec, pub_key_map, auth_cache) fetch_and_handle_events(db, origin, &state_vec, pub_key_map)
.await .await
.map_err(|_| "Failed to fetch state events locally".to_owned())? .map_err(|_| "Failed to fetch state events locally".to_owned())?
.into_iter() .into_iter()
@ -1069,18 +1057,12 @@ pub fn handle_incoming_pdu<'a>(
{ {
Ok(res) => { Ok(res) => {
debug!("Fetching state events at event."); debug!("Fetching state events at event.");
let state_vec = match fetch_and_handle_events( let state_vec =
&db, match fetch_and_handle_events(&db, origin, &res.pdu_ids, pub_key_map).await
origin, {
&res.pdu_ids, Ok(state) => state,
pub_key_map, Err(_) => return Err("Failed to fetch state events.".to_owned()),
auth_cache, };
)
.await
{
Ok(state) => state,
Err(_) => return Err("Failed to fetch state events.".to_owned()),
};
let mut state = BTreeMap::new(); let mut state = BTreeMap::new();
for pdu in state_vec { for pdu in state_vec {
@ -1106,14 +1088,8 @@ pub fn handle_incoming_pdu<'a>(
} }
debug!("Fetching auth chain events at event."); debug!("Fetching auth chain events at event.");
incoming_auth_events = match fetch_and_handle_events( match fetch_and_handle_events(&db, origin, &res.auth_chain_ids, pub_key_map)
&db, .await
origin,
&res.auth_chain_ids,
pub_key_map,
auth_cache,
)
.await
{ {
Ok(state) => state, Ok(state) => state,
Err(_) => return Err("Failed to fetch auth chain.".to_owned()), Err(_) => return Err("Failed to fetch auth chain.".to_owned()),
@ -1181,15 +1157,12 @@ pub fn handle_incoming_pdu<'a>(
let mut leaf_state = db let mut leaf_state = db
.rooms .rooms
.state_full(pdu_shortstatehash) .state_full(pdu_shortstatehash)
.map_err(|_| "Failed to ask db for room state.".to_owned())? .map_err(|_| "Failed to ask db for room state.".to_owned())?;
.into_iter()
.map(|(k, v)| (k, Arc::new(v)))
.collect::<StateMap<_>>();
if let Some(state_key) = &leaf_pdu.state_key { if let Some(state_key) = &leaf_pdu.state_key {
// Now it's the state after // Now it's the state after
let key = (leaf_pdu.kind.clone(), state_key.clone()); let key = (leaf_pdu.kind.clone(), state_key.clone());
leaf_state.insert(key, Arc::new(leaf_pdu)); leaf_state.insert(key, leaf_pdu);
} }
fork_states.insert(leaf_state); fork_states.insert(leaf_state);
@ -1209,10 +1182,7 @@ pub fn handle_incoming_pdu<'a>(
let current_state = db let current_state = db
.rooms .rooms
.room_state_full(&room_id) .room_state_full(&room_id)
.map_err(|_| "Failed to load room state.".to_owned())? .map_err(|_| "Failed to load room state.".to_owned())?;
.into_iter()
.map(|(k, v)| (k, Arc::new(v)))
.collect::<BTreeMap<_, _>>();
fork_states.insert(current_state.clone()); fork_states.insert(current_state.clone());
@ -1249,14 +1219,8 @@ pub fn handle_incoming_pdu<'a>(
for map in &fork_states { for map in &fork_states {
let mut state_auth = vec![]; let mut state_auth = vec![];
for auth_id in map.values().flat_map(|pdu| &pdu.auth_events) { for auth_id in map.values().flat_map(|pdu| &pdu.auth_events) {
match fetch_and_handle_events( match fetch_and_handle_events(&db, origin, &[auth_id.clone()], pub_key_map)
&db, .await
origin,
&[auth_id.clone()],
pub_key_map,
auth_cache,
)
.await
{ {
// This should always contain exactly one element when Ok // This should always contain exactly one element when Ok
Ok(events) => state_auth.extend_from_slice(&events), Ok(events) => state_auth.extend_from_slice(&events),
@ -1265,31 +1229,9 @@ pub fn handle_incoming_pdu<'a>(
} }
} }
} }
auth_cache.extend(
map.iter()
.map(|pdu| (pdu.1.event_id.clone(), pdu.1.clone())),
);
auth_events.push(state_auth); auth_events.push(state_auth);
} }
// Add everything we will need to event_map
auth_cache.extend(
auth_events
.iter()
.map(|pdus| pdus.iter().map(|pdu| (pdu.event_id.clone(), pdu.clone())))
.flatten(),
);
auth_cache.extend(
incoming_auth_events
.into_iter()
.map(|pdu| (pdu.event_id().clone(), pdu)),
);
auth_cache.extend(
state_after
.into_iter()
.map(|(_, pdu)| (pdu.event_id().clone(), pdu)),
);
match state_res::StateResolution::resolve( match state_res::StateResolution::resolve(
&room_id, &room_id,
room_version_id, room_version_id,
@ -1305,7 +1247,13 @@ pub fn handle_incoming_pdu<'a>(
.into_iter() .into_iter()
.map(|pdus| pdus.into_iter().map(|pdu| pdu.event_id().clone()).collect()) .map(|pdus| pdus.into_iter().map(|pdu| pdu.event_id().clone()).collect())
.collect(), .collect(),
auth_cache, &|id| {
let res = db.rooms.get_pdu(id);
if let Err(e) = &res {
error!("LOOK AT ME Failed to fetch event: {}", e);
}
res.ok().flatten()
},
) { ) {
Ok(new_state) => new_state, Ok(new_state) => new_state,
Err(_) => { Err(_) => {
@ -1365,21 +1313,16 @@ pub fn handle_incoming_pdu<'a>(
/// Find the event and auth it. Once the event is validated (steps 1 - 8) /// Find the event and auth it. Once the event is validated (steps 1 - 8)
/// it is appended to the outliers Tree. /// it is appended to the outliers Tree.
/// ///
/// a. Look in the auth_cache /// a. Look in the main timeline (pduid_pdu tree)
/// b. Look in the main timeline (pduid_pdu tree) /// b. Look at outlier pdu tree
/// c. Look at outlier pdu tree /// c. Ask origin server over federation
/// d. Ask origin server over federation /// d. TODO: Ask other servers over federation?
/// e. TODO: Ask other servers over federation?
///
/// If the event is unknown to the `auth_cache` it is added. This guarantees that any
/// event we need to know of will be present.
//#[tracing::instrument(skip(db, key_map, auth_cache))] //#[tracing::instrument(skip(db, key_map, auth_cache))]
pub(crate) fn fetch_and_handle_events<'a>( pub(crate) fn fetch_and_handle_events<'a>(
db: &'a Database, db: &'a Database,
origin: &'a ServerName, origin: &'a ServerName,
events: &'a [EventId], events: &'a [EventId],
pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, String>>>, pub_key_map: &'a RwLock<BTreeMap<String, BTreeMap<String, String>>>,
auth_cache: &'a mut EventMap<Arc<PduEvent>>,
) -> AsyncRecursiveResult<'a, Vec<Arc<PduEvent>>, Error> { ) -> AsyncRecursiveResult<'a, Vec<Arc<PduEvent>>, Error> {
Box::pin(async move { Box::pin(async move {
let back_off = |id| match db.globals.bad_event_ratelimiter.write().unwrap().entry(id) { let back_off = |id| match db.globals.bad_event_ratelimiter.write().unwrap().entry(id) {
@ -1403,84 +1346,73 @@ pub(crate) fn fetch_and_handle_events<'a>(
continue; continue;
} }
} }
// a. Look at auth cache
let pdu = match auth_cache.get(id) { // a. Look in the main timeline (pduid_pdu tree)
// b. Look at outlier pdu tree
// (get_pdu checks both)
let pdu = match db.rooms.get_pdu(&id)? {
Some(pdu) => { Some(pdu) => {
// We already have the auth chain for events in cache trace!("Found {} in db", id);
pdu.clone() pdu
} }
// b. Look in the main timeline (pduid_pdu tree) None => {
// c. Look at outlier pdu tree // c. Ask origin server over federation
// (get_pdu checks both) debug!("Fetching {} over federation.", id);
None => match db.rooms.get_pdu(&id)? { match db
Some(pdu) => { .sending
trace!("Found {} in db", id); .send_federation_request(
// We need to fetch the auth chain &db.globals,
let _ = fetch_and_handle_events(
db,
origin, origin,
&pdu.auth_events, get_event::v1::Request { event_id: &id },
pub_key_map,
auth_cache,
) )
.await?; .await
Arc::new(pdu) {
} Ok(res) => {
None => { debug!("Got {} over federation", id);
// d. Ask origin server over federation let (event_id, mut value) =
debug!("Fetching {} over federation.", id); crate::pdu::gen_event_id_canonical_json(&res.pdu)?;
match db // This will also fetch the auth chain
.sending match handle_incoming_pdu(
.send_federation_request(
&db.globals,
origin, origin,
get_event::v1::Request { event_id: &id }, &event_id,
value.clone(),
false,
db,
pub_key_map,
) )
.await .await
{ {
Ok(res) => { Ok(_) => {
debug!("Got {} over federation", id); value.insert(
let (event_id, mut value) = "event_id".to_owned(),
crate::pdu::gen_event_id_canonical_json(&res.pdu)?; CanonicalJsonValue::String(event_id.into()),
// This will also fetch the auth chain );
match handle_incoming_pdu(
origin,
&event_id,
value.clone(),
false,
db,
pub_key_map,
auth_cache,
)
.await
{
Ok(_) => {
value.insert(
"event_id".to_owned(),
CanonicalJsonValue::String(event_id.into()),
);
Arc::new(serde_json::from_value( Arc::new(
serde_json::to_value(value).expect("canonicaljsonobject is valid value"), serde_json::from_value(
).expect("This is possible because handle_incoming_pdu worked")) serde_json::to_value(value)
} .expect("canonicaljsonobject is valid value"),
Err(e) => { )
warn!("Authentication of event {} failed: {:?}", id, e); .expect(
back_off(id.clone()); "This is possible because handle_incoming_pdu worked",
continue; ),
} )
}
Err(e) => {
warn!("Authentication of event {} failed: {:?}", id, e);
back_off(id.clone());
continue;
} }
} }
Err(_) => { }
warn!("Failed to fetch event: {}", id); Err(_) => {
back_off(id.clone()); warn!("Failed to fetch event: {}", id);
continue; back_off(id.clone());
} continue;
} }
} }
}, }
}; };
auth_cache.entry(id.clone()).or_insert_with(|| pdu.clone());
pdus.push(pdu); pdus.push(pdu);
} }
Ok(pdus) Ok(pdus)
@ -1838,7 +1770,7 @@ pub fn get_event_authorization_route(
.difference(&auth_chain_ids) .difference(&auth_chain_ids)
.cloned(), .cloned(),
); );
auth_chain_ids.extend(pdu.auth_events.into_iter()); auth_chain_ids.extend(pdu.auth_events.clone().into_iter());
let pdu_json = PduEvent::convert_to_outgoing_federation_event( let pdu_json = PduEvent::convert_to_outgoing_federation_event(
db.rooms.get_pdu_json(&event_id)?.unwrap(), db.rooms.get_pdu_json(&event_id)?.unwrap(),
@ -1901,7 +1833,7 @@ pub fn get_room_state_route(
.difference(&auth_chain_ids) .difference(&auth_chain_ids)
.cloned(), .cloned(),
); );
auth_chain_ids.extend(pdu.auth_events.into_iter()); auth_chain_ids.extend(pdu.auth_events.clone().into_iter());
let pdu_json = PduEvent::convert_to_outgoing_federation_event( let pdu_json = PduEvent::convert_to_outgoing_federation_event(
db.rooms.get_pdu_json(&event_id)?.unwrap(), db.rooms.get_pdu_json(&event_id)?.unwrap(),
@ -1954,7 +1886,7 @@ pub fn get_room_state_ids_route(
.difference(&auth_chain_ids) .difference(&auth_chain_ids)
.cloned(), .cloned(),
); );
auth_chain_ids.extend(pdu.auth_events.into_iter()); auth_chain_ids.extend(pdu.auth_events.clone().into_iter());
} else { } else {
warn!("Could not find pdu mentioned in auth events."); warn!("Could not find pdu mentioned in auth events.");
} }
@ -2022,7 +1954,7 @@ pub fn create_join_event_template_route(
let create_prev_event = if prev_events.len() == 1 let create_prev_event = if prev_events.len() == 1
&& Some(&prev_events[0]) == create_event.as_ref().map(|c| &c.event_id) && Some(&prev_events[0]) == create_event.as_ref().map(|c| &c.event_id)
{ {
create_event.map(Arc::new) create_event
} else { } else {
None None
}; };
@ -2066,10 +1998,10 @@ pub fn create_join_event_template_route(
let mut unsigned = BTreeMap::new(); let mut unsigned = BTreeMap::new();
if let Some(prev_pdu) = db.rooms.room_state_get(&body.room_id, &kind, &state_key)? { if let Some(prev_pdu) = db.rooms.room_state_get(&body.room_id, &kind, &state_key)? {
unsigned.insert("prev_content".to_owned(), prev_pdu.content); unsigned.insert("prev_content".to_owned(), prev_pdu.content.clone());
unsigned.insert( unsigned.insert(
"prev_sender".to_owned(), "prev_sender".to_owned(),
serde_json::to_value(prev_pdu.sender).expect("UserId::to_value always works"), serde_json::to_value(&prev_pdu.sender).expect("UserId::to_value always works"),
); );
} }
@ -2161,7 +2093,7 @@ pub async fn create_join_event_route(
))?; ))?;
let pub_key_map = RwLock::new(BTreeMap::new()); let pub_key_map = RwLock::new(BTreeMap::new());
let mut auth_cache = EventMap::new(); // let mut auth_cache = EventMap::new();
// We do not add the event_id field to the pdu here because of signature and hashes checks // We do not add the event_id field to the pdu here because of signature and hashes checks
let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&body.pdu) { let (event_id, value) = match crate::pdu::gen_event_id_canonical_json(&body.pdu) {
@ -2184,26 +2116,18 @@ pub async fn create_join_event_route(
) )
.map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Origin field is invalid."))?; .map_err(|_| Error::BadRequest(ErrorKind::InvalidParam, "Origin field is invalid."))?;
let pdu_id = handle_incoming_pdu( let pdu_id = handle_incoming_pdu(&origin, &event_id, value, true, &db, &pub_key_map)
&origin, .await
&event_id, .map_err(|_| {
value, Error::BadRequest(
true, ErrorKind::InvalidParam,
&db, "Error while handling incoming PDU.",
&pub_key_map, )
&mut auth_cache, })?
) .ok_or(Error::BadRequest(
.await
.map_err(|_| {
Error::BadRequest(
ErrorKind::InvalidParam, ErrorKind::InvalidParam,
"Error while handling incoming PDU.", "Could not accept incoming PDU as timeline event.",
) ))?;
})?
.ok_or(Error::BadRequest(
ErrorKind::InvalidParam,
"Could not accept incoming PDU as timeline event.",
))?;
let state_ids = db.rooms.state_full_ids(shortstatehash)?; let state_ids = db.rooms.state_full_ids(shortstatehash)?;
@ -2220,7 +2144,7 @@ pub async fn create_join_event_route(
.difference(&auth_chain_ids) .difference(&auth_chain_ids)
.cloned(), .cloned(),
); );
auth_chain_ids.extend(pdu.auth_events.into_iter()); auth_chain_ids.extend(pdu.auth_events.clone().into_iter());
} else { } else {
warn!("Could not find pdu mentioned in auth events."); warn!("Could not find pdu mentioned in auth events.");
} }