From 0b13fa9ffbdc2258b868eaed83fb844e90d3dd83 Mon Sep 17 00:00:00 2001 From: Ginger Date: Sat, 4 Jul 2026 19:27:53 -0400 Subject: [PATCH] fix: Adjust sync watcher logic to be more explicit --- changelog.d/+32bb7e0d.bugfix.md | 1 + src/api/client/membership/leave.rs | 6 +- src/api/client/read_marker.rs | 4 + src/api/client/sync/v3/mod.rs | 5 +- src/api/client/sync/v5.rs | 5 +- src/api/server/invite.rs | 2 + src/service/account_data/mod.rs | 6 +- .../event_handler/handle_incoming_pdu.rs | 4 + src/service/rooms/event_handler/mod.rs | 4 +- src/service/rooms/membership/mod.rs | 6 +- src/service/rooms/read_receipt/mod.rs | 5 +- src/service/rooms/state/mod.rs | 39 +++--- src/service/rooms/state_cache/mod.rs | 6 +- src/service/rooms/state_cache/update.rs | 40 +++--- src/service/rooms/timeline/append.rs | 22 ++-- src/service/rooms/timeline/build.rs | 2 + src/service/rooms/timeline/mod.rs | 51 ++++---- src/service/sync/mod.rs | 96 +++++++-------- src/service/sync/watch.rs | 115 ------------------ src/service/users/device.rs | 4 + src/service/users/keys.rs | 11 +- src/service/users/mod.rs | 4 +- 22 files changed, 189 insertions(+), 249 deletions(-) create mode 100644 changelog.d/+32bb7e0d.bugfix.md delete mode 100644 src/service/sync/watch.rs diff --git a/changelog.d/+32bb7e0d.bugfix.md b/changelog.d/+32bb7e0d.bugfix.md new file mode 100644 index 000000000..51dfa9efe --- /dev/null +++ b/changelog.d/+32bb7e0d.bugfix.md @@ -0,0 +1 @@ +Improved invite and join reliability in clients using legacy sync. Contributed by @ginger diff --git a/src/api/client/membership/leave.rs b/src/api/client/membership/leave.rs index d4c9add00..afb9e0e97 100644 --- a/src/api/client/membership/leave.rs +++ b/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(()) } diff --git a/src/api/client/read_marker.rs b/src/api/client/read_marker.rs index e1e66764a..a476f6790 100644 --- a/src/api/client/read_marker.rs +++ b/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()) } diff --git a/src/api/client/sync/v3/mod.rs b/src/api/client/sync/v3/mod.rs index 9be5cb895..90b8ca83b 100644 --- a/src/api/client/sync/v3/mod.rs +++ b/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 diff --git a/src/api/client/sync/v5.rs b/src/api/client/sync/v5.rs index ea1f1a00a..3602cc45b 100644 --- a/src/api/client/sync/v5.rs +++ b/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?; diff --git a/src/api/server/invite.rs b/src/api/server/invite.rs index a029b822c..9d2fab0e3 100644 --- a/src/api/server/invite.rs +++ b/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 diff --git a/src/service/account_data/mod.rs b/src/service/account_data/mod.rs index 5622826a5..52e9cb9b2 100644 --- a/src/service/account_data/mod.rs +++ b/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, + sync: Dep, } impl crate::Service for Service { @@ -43,6 +44,7 @@ fn build(args: crate::Args<'_>) -> Result> { Ok(Arc::new(Self { services: Services { globals: args.depend::("globals"), + sync: args.depend::("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(()) } diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs index f9a497f3c..86be9c44b 100644 --- a/src/service/rooms/event_handler/handle_incoming_pdu.rs +++ b/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); } } diff --git a/src/service/rooms/event_handler/mod.rs b/src/service/rooms/event_handler/mod.rs index 3eae2f477..779889287 100644 --- a/src/service/rooms/event_handler/mod.rs +++ b/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, @@ -44,6 +44,7 @@ struct Services { state_cache: Dep, state_accessor: Dep, state_compressor: Dep, + sync: Dep, timeline: Dep, server: Arc, } @@ -74,6 +75,7 @@ fn build(args: crate::Args<'_>) -> Result> { .depend::("rooms::state_accessor"), state_compressor: args .depend::("rooms::state_compressor"), + sync: args.depend::("sync"), timeline: args.depend::("rooms::timeline"), server: args.server.clone(), }, diff --git a/src/service/rooms/membership/mod.rs b/src/service/rooms/membership/mod.rs index 49271bded..02e0a8d26 100644 --- a/src/service/rooms/membership/mod.rs +++ b/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_cache: Dep, state_compressor: Dep, + sync: Dep, timeline: Dep, users: Dep, } @@ -87,6 +88,7 @@ fn build(args: crate::Args<'_>) -> Result> { state_cache: args.depend::("rooms::state_cache"), state_compressor: args .depend::("rooms::state_compressor"), + sync: args.depend::("sync"), timeline: args.depend::("rooms::timeline"), users: args.depend::("users"), }, @@ -672,6 +674,8 @@ pub async fn join_remote_room( } drop(cork); + self.services.sync.wake_all_joined(room_id).await; + Ok(()) } diff --git a/src/service/rooms/read_receipt/mod.rs b/src/service/rooms/read_receipt/mod.rs index 89280a34e..1eda4aa48 100644 --- a/src/service/rooms/read_receipt/mod.rs +++ b/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, short: Dep, + sync: Dep, timeline: Dep, } @@ -40,6 +41,7 @@ fn build(args: crate::Args<'_>) -> Result> { services: Services { sending: args.depend::("sending"), short: args.depend::("rooms::short"), + sync: args.depend::("sync"), timeline: args.depend::("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 diff --git a/src/service/rooms/state/mod.rs b/src/service/rooms/state/mod.rs index bb0061f08..09db154f2 100644 --- a/src/service/rooms/state/mod.rs +++ b/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, state_accessor: Dep, state_compressor: Dep, + sync: Dep, timeline: Dep, } @@ -68,6 +71,7 @@ fn build(args: crate::Args<'_>) -> Result> { .depend::("rooms::state_accessor"), state_compressor: args .depend::("rooms::state_compressor"), + sync: args.depend::("sync"), timeline: args.depend::("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 - where - E: Event + Send + Sync, - &'a E: Event + Send, - { + ) -> Vec { 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 diff --git a/src/service/rooms/state_cache/mod.rs b/src/service/rooms/state_cache/mod.rs index 5c64ce04c..40464049a 100644 --- a/src/service/rooms/state_cache/mod.rs +++ b/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, state: Dep, state_accessor: Dep, + sync: Dep, users: Dep, } @@ -67,6 +70,7 @@ fn build(args: crate::Args<'_>) -> Result> { state: args.depend::("rooms::state"), state_accessor: args .depend::("rooms::state_accessor"), + sync: args.depend::("sync"), users: args.depend::("users"), }, db: Data { diff --git a/src/service/rooms/state_cache/update.rs b/src/service/rooms/state_cache/update.rs index 28f5098b7..197c4bc0f 100644 --- a/src/service/rooms/state_cache/update.rs +++ b/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::()?; + 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(()) } diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 132ee4f0a..3d7dcbcc1 100644 --- a/src/service/rooms/timeline/append.rs +++ b/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; diff --git a/src/service/rooms/timeline/build.rs b/src/service/rooms/timeline/build.rs index 8ff04b024..1c0ee0f94 100644 --- a/src/service/rooms/timeline/build.rs +++ b/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 = self .services .state_cache diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs index f4cae2ec7..c96707b43 100644 --- a/src/service/rooms/timeline/mod.rs +++ b/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, admin: Dep, alias: Dep, + event_handler: Dep, globals: Dep, - short: Dep, - state: Dep, - state_cache: Dep, - state_accessor: Dep, pdu_metadata: Dep, + pusher: Dep, read_receipt: Dep, + search: Dep, sending: Dep, server_keys: Dep, + short: Dep, + state: Dep, + state_accessor: Dep, + state_cache: Dep, + sync: Dep, + threads: Dep, user: Dep, users: Dep, - pusher: Dep, - threads: Dep, - search: Dep, - event_handler: Dep, } type RoomMutexMap = MutexMap; @@ -95,23 +97,24 @@ fn build(args: crate::Args<'_>) -> Result> { appservice: args.depend::("appservice"), admin: args.depend::("admin"), alias: args.depend::("rooms::alias"), - globals: args.depend::("globals"), - short: args.depend::("rooms::short"), - state: args.depend::("rooms::state"), - state_cache: args.depend::("rooms::state_cache"), - state_accessor: args - .depend::("rooms::state_accessor"), - pdu_metadata: args.depend::("rooms::pdu_metadata"), - read_receipt: args.depend::("rooms::read_receipt"), - sending: args.depend::("sending"), - server_keys: args.depend::("server_keys"), - user: args.depend::("rooms::user"), - users: args.depend::("users"), - pusher: args.depend::("pusher"), - threads: args.depend::("rooms::threads"), - search: args.depend::("rooms::search"), event_handler: args .depend::("rooms::event_handler"), + globals: args.depend::("globals"), + pdu_metadata: args.depend::("rooms::pdu_metadata"), + pusher: args.depend::("pusher"), + read_receipt: args.depend::("rooms::read_receipt"), + search: args.depend::("rooms::search"), + sending: args.depend::("sending"), + server_keys: args.depend::("server_keys"), + short: args.depend::("rooms::short"), + state: args.depend::("rooms::state"), + state_accessor: args + .depend::("rooms::state_accessor"), + state_cache: args.depend::("rooms::state_cache"), + sync: args.depend::("sync"), + threads: args.depend::("rooms::threads"), + user: args.depend::("rooms::user"), + users: args.depend::("users"), }, db: Data::new(&args), mutex_insert: RoomMutexMap::new(), diff --git a/src/service/sync/mod.rs b/src/service/sync/mod.rs index 44fc3ac2f..ff6812005 100644 --- a/src/service/sync/mod.rs +++ b/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, + wakers: Mutex>>, snake_connections: DbConnections, } -pub struct Data { - todeviceid_events: Arc, - userroomid_joined: Arc, - userroomid_invitestate: Arc, - userroomid_leftstate: Arc, - userroomid_notificationcount: Arc, - userroomid_highlightcount: Arc, - pduid_pdu: Arc, - keychangeid_userid: Arc, - roomusertype_roomuserdataid: Arc, - readreceiptid_readreceipt: Arc, - userid_lastonetimekeyupdate: Arc, -} - struct Services { - server: Arc, - short: Dep, state_cache: Dep, - typing: Dep, } #[allow(unused, reason = "TODO refactor")] @@ -58,33 +42,16 @@ struct SnakeSyncCache { type DbConnections = SyncMutex>; type DbConnectionsKey = (OwnedUserId, OwnedDeviceId, String); -type DbConnectionsVal = Arc>; type SnakeConnectionsKey = (OwnedUserId, OwnedDeviceId, Option); type SnakeConnectionsVal = Arc>; impl crate::Service for Service { fn build(args: crate::Args<'_>) -> Result> { 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"), state_cache: args.depend::("rooms::state_cache"), - typing: args.depend::("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 { + 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, diff --git a/src/service/sync/watch.rs b/src/service/sync/watch.rs deleted file mode 100644 index 3795b4338..000000000 --- a/src/service/sync/watch.rs +++ /dev/null @@ -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(()) - } -} diff --git a/src/service/users/device.rs b/src/service/users/device.rs index 9d436b152..a2bc33336 100644 --- a/src/service/users/device.rs +++ b/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( self.db.todeviceid_events.del(key); }) .await; + + self.services.sync.wake(user_id).await; } /// Updates device metadata and increments the device list version. diff --git a/src/service/users/keys.rs b/src/service/users/keys.rs index b6699f88c..0626a6cd3 100644 --- a/src/service/users/keys.rs +++ b/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; diff --git a/src/service/users/mod.rs b/src/service/users/mod.rs index 155cdeed4..f784ae878 100644 --- a/src/service/users/mod.rs +++ b/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, state_accessor: Dep, state_cache: Dep, + sync: Dep, threepid: Dep, timeline: Dep, } @@ -119,6 +120,7 @@ fn build(args: crate::Args<'_>) -> Result> { state_accessor: args .depend::("rooms::state_accessor"), state_cache: args.depend::("rooms::state_cache"), + sync: args.depend::("sync"), threepid: args.depend::("threepid"), timeline: args.depend::("rooms::timeline"), },