fix: Properly sync left rooms

- Remove most usages of `update_membership` in favor
  of directly calling the `mark_as_*` functions
- Store the leave membership event as the value in the
  `userroomid_leftstate` table
- Use the `userroomid_leftstate` table to synchronize the
  timeline and state for left rooms if possible
This commit is contained in:
Ginger
2025-12-07 19:58:24 +00:00
parent fceaaedc04
commit 1c21e4af6e
14 changed files with 759 additions and 665 deletions
+5 -1
View File
@@ -456,7 +456,11 @@ async fn retroactively_fix_bad_data_from_roomuserid_joined(services: &Services)
for user_id in &non_joined_members {
debug_info!("User is left or banned, marking as left");
services.rooms.state_cache.mark_as_left(user_id, room_id);
services
.rooms
.state_cache
.mark_as_left(user_id, room_id, None)
.await;
}
}
+2 -14
View File
@@ -20,7 +20,7 @@ use ruma::{
EventId, OwnedEventId, OwnedRoomId, RoomId, RoomVersionId, UserId,
events::{
AnyStrippedStateEvent, StateEventType, TimelineEventType,
room::{create::RoomCreateEventContent, member::RoomMemberEventContent},
room::create::RoomCreateEventContent,
},
serde::Raw,
};
@@ -126,21 +126,9 @@ impl Service {
continue;
};
let Ok(membership_event) = pdu.get_content::<RoomMemberEventContent>() else {
continue;
};
self.services
.state_cache
.update_membership(
room_id,
user_id,
membership_event,
&pdu.sender,
None,
None,
false,
)
.update_membership(room_id, user_id, &pdu, false)
.await?;
},
| TimelineEventType::SpaceChild => {
+15 -13
View File
@@ -4,7 +4,7 @@ mod via;
use std::{collections::HashMap, sync::Arc};
use conduwuit::{
Result, SyncRwLock, implement,
Pdu, Result, SyncRwLock, implement,
result::LogErr,
utils::{ReadyExt, stream::TryIgnore},
warn,
@@ -13,7 +13,7 @@ use database::{Deserialized, Ignore, Interfix, Map};
use futures::{Stream, StreamExt, future::join5, pin_mut};
use ruma::{
OwnedRoomId, OwnedUserId, RoomId, ServerName, UserId,
events::{AnyStrippedStateEvent, AnySyncStateEvent, room::member::MembershipState},
events::{AnyStrippedStateEvent, room::member::MembershipState},
serde::Raw,
};
@@ -54,7 +54,6 @@ struct Data {
type AppServiceInRoomCache = SyncRwLock<HashMap<OwnedRoomId, HashMap<String, bool>>>;
type StrippedStateEventItem = (OwnedRoomId, Vec<Raw<AnyStrippedStateEvent>>);
type SyncStateEventItem = (OwnedRoomId, Vec<Raw<AnySyncStateEvent>>);
impl crate::Service for Service {
fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
@@ -431,18 +430,16 @@ pub async fn knock_state(
#[implement(Service)]
#[tracing::instrument(skip(self), level = "trace")]
pub async fn left_state(
&self,
user_id: &UserId,
room_id: &RoomId,
) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
pub async fn left_state(&self, user_id: &UserId, room_id: &RoomId) -> Option<Pdu> {
let key = (user_id, room_id);
self.db
.userroomid_leftstate
.qry(&key)
.await
.deserialized()
.and_then(|val: Raw<Vec<AnyStrippedStateEvent>>| val.deserialize_as().map_err(Into::into))
// old databases may have garbage data as values in the `userroomid_leftstate` table from before
// the leave event was stored there. they still need to be included, so we return Ok(None) for deserialization failures.
.unwrap_or(None)
}
/// Returns an iterator over all rooms a user left.
@@ -451,8 +448,8 @@ pub async fn left_state(
pub fn rooms_left<'a>(
&'a self,
user_id: &'a UserId,
) -> impl Stream<Item = SyncStateEventItem> + Send + 'a {
type KeyVal<'a> = (Key<'a>, Raw<Vec<Raw<AnySyncStateEvent>>>);
) -> impl Stream<Item = (OwnedRoomId, Option<Pdu>)> + Send + 'a {
type KeyVal<'a> = (Key<'a>, Raw<Option<Pdu>>);
type Key<'a> = (&'a UserId, &'a RoomId);
let prefix = (user_id, Interfix);
@@ -461,8 +458,13 @@ pub fn rooms_left<'a>(
.stream_prefix(&prefix)
.ignore_err()
.map(|((_, room_id), state): KeyVal<'_>| (room_id.to_owned(), state))
.map(|(room_id, state)| Ok((room_id, state.deserialize_as()?)))
.ignore_err()
.ready_filter_map(|(room_id, state)| {
// deserialization errors need to be ignored. see comment in `left_state`
match state.deserialize() {
| Ok(state) => Some((room_id, state)),
| Err(_) => Some((room_id, None)),
}
})
}
#[implement(Service)]
+31 -30
View File
@@ -1,13 +1,13 @@
use std::collections::HashSet;
use conduwuit::{Err, Result, implement, is_not_empty, utils::ReadyExt, warn};
use conduwuit::{Err, Event, Pdu, Result, implement, is_not_empty, utils::ReadyExt, warn};
use database::{Json, serialize_key};
use futures::StreamExt;
use ruma::{
OwnedServerName, RoomId, UserId,
events::{
AnyStrippedStateEvent, AnySyncStateEvent, GlobalAccountDataEventType,
RoomAccountDataEventType, StateEventType,
AnyStrippedStateEvent, GlobalAccountDataEventType, RoomAccountDataEventType,
StateEventType,
direct::DirectEvent,
invite_permission_config::FilterLevel,
room::{
@@ -26,8 +26,7 @@ use ruma::{
fields(
%room_id,
%user_id,
%sender,
?membership_event,
?pdu,
),
)]
#[allow(clippy::too_many_arguments)]
@@ -35,13 +34,10 @@ pub async fn update_membership(
&self,
room_id: &RoomId,
user_id: &UserId,
membership_event: RoomMemberEventContent,
sender: &UserId,
last_state: Option<Vec<Raw<AnyStrippedStateEvent>>>,
invite_via: Option<Vec<OwnedServerName>>,
pdu: &Pdu,
update_joined_count: bool,
) -> Result {
let membership = membership_event.membership;
let membership = pdu.get_content::<RoomMemberEventContent>()?;
// Keep track what remote users exist by adding them as "deactivated" users
//
@@ -54,7 +50,7 @@ pub async fn update_membership(
}
}
match &membership {
match &membership.membership {
| MembershipState::Join => {
// Check if the user never joined this room
if !self.once_joined(user_id, room_id).await {
@@ -125,6 +121,7 @@ pub async fn update_membership(
// return an error for blocked invites. ignored invites aren't handled here
// since the recipient's membership should still be changed to `invite`.
// they're filtered out in the individual /sync handlers
let sender = pdu.sender();
if matches!(
self.services
.users
@@ -136,19 +133,15 @@ pub async fn update_membership(
"{user_id} has blocked invites from {sender}."
)));
}
self.mark_as_invited(user_id, room_id, sender, last_state, invite_via)
// TODO: make sure that passing None for `last_state` is correct behavior.
// the call from `append_pdu` used to use `services.state.summary_stripped`
// to fill that parameter.
self.mark_as_invited(user_id, room_id, sender, None, None)
.await;
},
| MembershipState::Leave | MembershipState::Ban => {
self.mark_as_left(user_id, room_id);
if self.services.globals.user_is_local(user_id)
&& (self.services.config.forget_forced_upon_leave
|| self.services.metadata.is_banned(room_id).await
|| self.services.metadata.is_disabled(room_id).await)
{
self.forget(room_id, user_id);
}
self.mark_as_left(user_id, room_id, Some(pdu.clone())).await;
},
| _ => {},
}
@@ -252,24 +245,24 @@ pub fn mark_as_joined(&self, user_id: &UserId, room_id: &RoomId) {
self.db.roomid_inviteviaservers.remove(room_id);
}
/// Direct DB function to directly mark a user as left. It is not
/// recommended to use this directly. You most likely should use
/// `update_membership` instead
/// Mark a user as having left a room.
///
/// `leave_pdu` represents the m.room.member event which the user sent to leave
/// the room. If this is None, no event was actually sent, but we must still
/// behave as if the user is no longer in the room. This may occur, for example,
/// if the room being left has been server-banned by an administrator.
#[implement(super::Service)]
#[tracing::instrument(skip(self), level = "debug")]
pub fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId) {
pub async fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId, leave_pdu: Option<Pdu>) {
let userroom_id = (user_id, room_id);
let userroom_id = serialize_key(userroom_id).expect("failed to serialize userroom_id");
let roomuser_id = (room_id, user_id);
let roomuser_id = serialize_key(roomuser_id).expect("failed to serialize roomuser_id");
// (timo) TODO
let leftstate = Vec::<Raw<AnySyncStateEvent>>::new();
self.db
.userroomid_leftstate
.raw_put(&userroom_id, Json(leftstate));
.raw_put(&userroom_id, Json(leave_pdu));
self.db
.roomuserid_leftcount
.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
@@ -285,6 +278,14 @@ pub fn mark_as_left(&self, user_id: &UserId, room_id: &RoomId) {
self.db.roomuserid_knockedcount.remove(&roomuser_id);
self.db.roomid_inviteviaservers.remove(room_id);
if self.services.globals.user_is_local(user_id)
&& (self.services.config.forget_forced_upon_leave
|| self.services.metadata.is_banned(room_id).await
|| self.services.metadata.is_disabled(room_id).await)
{
self.forget(room_id, user_id);
}
}
/// Direct DB function to directly mark a user as knocked. It is not
@@ -366,7 +367,7 @@ pub async fn mark_as_invited(
.raw_aput::<8, _, _>(&roomuser_id, self.services.globals.next_count().unwrap());
self.db
.userroomid_invitesender
.raw_put(&userroom_id, sender_user);
.insert(&userroom_id, sender_user);
self.db.userroomid_joined.remove(&userroom_id);
self.db.roomuserid_joined.remove(&roomuser_id);
+2 -23
View File
@@ -19,9 +19,7 @@ use ruma::{
GlobalAccountDataEventType, StateEventType, TimelineEventType,
push_rules::PushRulesEvent,
room::{
encrypted::Relation,
member::{MembershipState, RoomMemberEventContent},
power_levels::RoomPowerLevelsEventContent,
encrypted::Relation, power_levels::RoomPowerLevelsEventContent,
redaction::RoomRedactionEventContent,
},
},
@@ -323,31 +321,12 @@ where
let target_user_id =
UserId::parse(state_key).expect("This state_key was previously validated");
let content: RoomMemberEventContent = pdu.get_content()?;
let stripped_state = match content.membership {
| MembershipState::Invite | MembershipState::Knock => self
.services
.state
.summary_stripped(pdu, room_id)
.await
.into(),
| _ => None,
};
// Update our membership info, we do this here incase a user is invited or
// knocked and immediately leaves we need the DB to record the invite or
// knock event for auth
self.services
.state_cache
.update_membership(
room_id,
target_user_id,
content,
pdu.sender(),
stripped_state,
None,
true,
)
.update_membership(room_id, target_user_id, &pdu, true)
.await?;
}
},