diff --git a/changelog.d/+account-data-replaced-entries.bugfix.md b/changelog.d/+account-data-replaced-entries.bugfix.md new file mode 100644 index 000000000..c3751d5a9 --- /dev/null +++ b/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. diff --git a/src/service/account_data/mod.rs b/src/service/account_data/mod.rs index 52e9cb9b2..14301ee91 100644 --- a/src/service/account_data/mod.rs +++ b/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, } 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 diff --git a/src/service/migrations.rs b/src/service/migrations.rs index 727bcce17..666994d9e 100644 --- a/src/service/migrations.rs +++ b/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, 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 = 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::(); + + 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();