fix: Keep one account data entry per type when updates race

Signed-off-by: Erwan Leboucher <erwanleboucher@gmail.com>

changelog.d/+account-data-replaced-entries.bugfix.md

@@ -0,0 +1 @@
+Fix account data updates sent at the same time leaving old copies behind, which every initial sync then returned. Existing copies are removed on startup. Contributed by @eleboucher.

src/service/account_data/mod.rs

@@ -2,7 +2,7 @@ use std::sync::Arc;
 
 use conduwuit::{
 	Err, Result, err,
-	utils::{ReadyExt, result::LogErr, stream::TryIgnore},
+	utils::{MutexMap, ReadyExt, result::LogErr, stream::TryIgnore},
 };
 use database::{Deserialized, Handle, Ignore, Json, Map};
 use futures::{Stream, StreamExt, TryFutureExt};
@@ -27,6 +27,7 @@ pub enum AnyRawAccountDataEvent {
 pub struct Service {
 	services: Services,
 	db: Data,
+	update_mutex: MutexMap<OwnedUserId, ()>,
 }
 
 struct Data {
@@ -50,6 +51,7 @@ impl crate::Service for Service {
 				roomuserdataid_accountdata: args.db["roomuserdataid_accountdata"].clone(),
 				roomusertype_roomuserdataid: args.db["roomusertype_roomuserdataid"].clone(),
 			},
+			update_mutex: MutexMap::new(),
 		}))
 	}
 
@@ -71,6 +73,10 @@ impl Service {
 			return Err!(Request(InvalidParam("Account data doesn't have all required fields.")));
 		}
 
+		// Two updates reading the same previous entry would both remove it,
+		// and the entry of the one that lost the index would stay forever.
+		let _lock = self.update_mutex.lock(user_id).await;
+
 		let count = self.services.globals.next_count().unwrap();
 		let roomuserdataid = (room_id, user_id, count, &event_type);
 		self.db

src/service/migrations.rs

@@ -18,7 +18,7 @@ use conduwuit::{
 	},
 	warn,
 };
-use database::Json;
+use database::{Deserialized, Ignore, Json};
 use futures::{FutureExt, StreamExt, TryStreamExt};
 use itertools::Itertools;
 use ruma::{
@@ -173,6 +173,9 @@ migration!(stringify!($name) => $name)
 		// Re-calculates extremities for any rooms that actually lack any
 		// extremities at all.
 		migration!(consistency_check_extremity_tracker),
+		// This is backwards compatible because old versions never read the
+		// account data entries it removes.
+		migration!(remove_replaced_account_data),
 	]
 };
 
@@ -1018,6 +1021,35 @@ async fn remove_unknown_extremities(services: &Services) -> Result {
 	Ok(())
 }
 
+async fn remove_replaced_account_data(services: &Services) -> Result {
+	type Key = (Option<OwnedRoomId>, OwnedUserId, u64, String);
+	type Latest = (Ignore, Ignore, u64, Ignore);
+
+	let entries = services.db["roomuserdataid_accountdata"].clone();
+	let latest_of_type = services.db["roomusertype_roomuserdataid"].clone();
+
+	let keys: Vec<Key> = entries.keys().ignore_err().collect().await;
+	let mut removed: usize = 0;
+	for (room_id, user_id, count, event_type) in &keys {
+		let latest = latest_of_type
+			.qry(&(room_id, user_id, event_type))
+			.await
+			.deserialized::<Latest>();
+
+		if latest.is_ok_and(|(_, _, latest, _)| latest != *count) {
+			entries.del((room_id, user_id, count, event_type));
+			removed = removed.saturating_add(1);
+		}
+	}
+
+	info!(
+		total = keys.len(),
+		?removed,
+		"Removed replaced entries in 'roomuserdataid_accountdata'."
+	);
+	Ok(())
+}
+
 #[tracing::instrument(skip_all)]
 async fn consistency_check_extremity_tracker(services: &Services) -> Result {
 	let mut rooms = services.rooms.metadata.iter_ids();