fix: Adjust sync watcher logic to be more explicit

changelog.d/+32bb7e0d.bugfix.md

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

src/api/client/membership/leave.rs

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

src/api/client/read_marker.rs

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

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

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

src/api/client/sync/v5.rs

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

src/api/server/invite.rs

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

src/service/account_data/mod.rs

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

src/service/rooms/event_handler/handle_incoming_pdu.rs

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

src/service/rooms/event_handler/mod.rs

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

src/service/rooms/membership/mod.rs

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

src/service/rooms/read_receipt/mod.rs

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

src/service/rooms/state/mod.rs

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

src/service/rooms/state_cache/mod.rs

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

src/service/rooms/state_cache/update.rs

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

src/service/rooms/timeline/append.rs

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

src/service/rooms/timeline/build.rs

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

src/service/rooms/timeline/mod.rs

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

src/service/sync/mod.rs

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

src/service/sync/watch.rs

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

src/service/users/device.rs

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

src/service/users/keys.rs

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

src/service/users/mod.rs

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