tuwunel_service/rooms/state_cache/
via.rs

1//! Routing-server hints derived from membership and invitation state.
2//!
3//! Power levels and joined-member distribution produce prospective routing
4//! servers, while inbound invite hints are kept in a compact aggregate row.
5//! The aggregate reader exposes only the final stored server from each row.
6
7use std::cmp::Reverse;
8
9use futures::{Stream, StreamExt, stream::iter};
10use ruma::{
11	OwnedServerName, RoomId, ServerName,
12	events::{StateEventType, room::power_levels::RoomPowerLevelsEventContent},
13	int,
14};
15use tuwunel_core::{
16	Result, implement,
17	itertools::Itertools,
18	utils::{StreamTools, stream::TryIgnore},
19	warn,
20};
21use tuwunel_database::{Ignore, Txn};
22
23/// Merges invitation routing hints into the caller's membership transaction.
24///
25/// The existing reader exposes only the aggregate row's final server, so older
26/// hints before that tail are not retained by this rewrite. Concurrent callers
27/// are not serialized and can overwrite one another's read-modify-write result.
28#[implement(super::Service)]
29#[tracing::instrument(level = "debug", skip(self, txn, servers))]
30pub(crate) async fn add_servers_invite_via(
31	&self,
32	txn: &mut Txn,
33	room_id: &RoomId,
34	servers: Vec<OwnedServerName>,
35) {
36	let servers = self
37		.servers_invite_via(room_id)
38		.map(ToOwned::to_owned)
39		.chain(iter(servers.into_iter()))
40		.collect::<Vec<_>>()
41		.await
42		.into_iter()
43		.sorted_unstable()
44		.dedup()
45		.collect_vec();
46
47	let servers = servers
48		.iter()
49		.map(|server| server.as_bytes())
50		.collect_vec()
51		.join(&[0xFF][..]);
52
53	txn.insert_raw(&self.db.roomid_inviteviaservers, room_id.as_bytes(), &servers);
54}
55
56/// Selects up to five servers likely to remain useful for room routing.
57///
58/// The highest-power user's server is considered first, followed by servers in
59/// descending joined-user count. The two sources are not deduplicated, and a
60/// missing power-level event simply omits the first candidate.
61///
62/// See <https://spec.matrix.org/latest/appendices/#routing>.
63#[implement(super::Service)]
64#[tracing::instrument(skip(self), level = "trace")]
65pub async fn servers_route_via(&self, room_id: &RoomId) -> Result<Vec<OwnedServerName>> {
66	let most_powerful = self.most_powerful_user_server(room_id).await;
67
68	Ok(most_powerful
69		.into_iter()
70		.chain(self.popular_servers(room_id).await)
71		.take(5)
72		.collect())
73}
74
75/// Returns the highest-power user's server when its level is at least 50.
76///
77/// Missing, unreadable, or malformed power-level state returns `None`.
78/// Equal-power ties follow the underlying users map's iteration order.
79#[implement(super::Service)]
80#[tracing::instrument(skip(self), level = "trace")]
81pub async fn most_powerful_user_server(&self, room_id: &RoomId) -> Option<OwnedServerName> {
82	self.services
83		.state_accessor
84		.room_state_get_content(room_id, &StateEventType::RoomPowerLevels, "")
85		.await
86		.ok()
87		.and_then(|content: RoomPowerLevelsEventContent| {
88			content
89				.users
90				.into_iter()
91				.max_by_key(|(_, power)| *power)
92				.filter(|(_, power)| *power >= int!(50))
93				.map(|(user, _)| user.server_name().to_owned())
94		})
95}
96
97/// Returns participating servers ordered by descending joined-user count.
98///
99/// Joined members are counted per server and the resulting servers are sorted.
100/// Read failures skipped by the membership stream can reduce the observed
101/// counts, and equal-count ordering is unspecified.
102#[implement(super::Service)]
103#[tracing::instrument(skip(self), level = "trace")]
104pub async fn popular_servers(&self, room_id: &RoomId) -> Vec<OwnedServerName> {
105	self.room_members(room_id)
106		.counts_by(|user| user.server_name().to_owned())
107		.await
108		.into_iter()
109		.sorted_by_key(|(_, users)| Reverse(*users))
110		.map(|(server, _)| server)
111		.collect()
112}
113
114/// Streams the final routing hint from each matching aggregate row.
115///
116/// The current room-keyed representation stores several servers in one value,
117/// but this accessor exposes only the last decoded server and therefore at most
118/// one item per room. Storage and decoding failures are skipped. Yielded names
119/// borrow the cursor and are invalid after the next poll; consume or own each
120/// item before advancing.
121#[implement(super::Service)]
122#[tracing::instrument(skip(self), level = "debug")]
123pub fn servers_invite_via<'a>(
124	&'a self,
125	room_id: &'a RoomId,
126) -> impl Stream<Item = &ServerName> + Send + 'a {
127	type KeyVal<'a> = (Ignore, Vec<&'a ServerName>);
128
129	self.db
130		.roomid_inviteviaservers
131		.stream_raw_prefix(room_id)
132		.ignore_err()
133		.map(|(_, servers): KeyVal<'_>| *servers.last().expect("at least one server"))
134}
tuwunel_service/rooms/state_cache/via.rs

tuwunel_service/rooms/state_cache/
via.rs
