From 18b0a87b2d3f7922f191b2a08258fb8073284f93 Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Thu, 20 Aug 2026 09:00:51 +0200 Subject: [PATCH] fix(sync): Order sliding sync rooms by activity --- .../sliding-sync-activity-ordering.bugfix.md | 1 + src/api/client/sync/v5.rs | 75 +++++++++++---- src/database/maps.rs | 4 + src/service/rooms/timeline/append.rs | 4 +- src/service/rooms/timeline/data.rs | 26 +++++ src/service/rooms/timeline/mod.rs | 5 + src/service/sync/mod.rs | 95 +++++++++++++++---- 7 files changed, 174 insertions(+), 36 deletions(-) create mode 100644 changelog.d/sliding-sync-activity-ordering.bugfix.md diff --git a/changelog.d/sliding-sync-activity-ordering.bugfix.md b/changelog.d/sliding-sync-activity-ordering.bugfix.md new file mode 100644 index 000000000..d3e6bf64b --- /dev/null +++ b/changelog.d/sliding-sync-activity-ordering.bugfix.md @@ -0,0 +1 @@ +Fixed simplified Sliding Sync lists returning rooms out of activity order. Contributed by @eleboucher. diff --git a/src/api/client/sync/v5.rs b/src/api/client/sync/v5.rs index 8ad9c6cc9..0b2894dde 100644 --- a/src/api/client/sync/v5.rs +++ b/src/api/client/sync/v5.rs @@ -124,7 +124,12 @@ pub(crate) async fn sync_events_v5_route( let snake_key = into_snake_key(sender_user, sender_device.as_str(), conn_id); - if globalsince != 0 && !services.sync.snake_connection_cached(&snake_key) { + if globalsince != 0 + && (!services.sync.snake_connection_cached(&snake_key) + || !services + .sync + .snake_connection_accepts_pos(&snake_key, globalsince)) + { return Err!(Request(UnknownPos( "Connection data unknown to server; restarting sync stream." ))); @@ -136,9 +141,10 @@ pub(crate) async fn sync_events_v5_route( } // Get sticky parameters from cache - let known_rooms = services - .sync - .update_snake_sync_request_with_cache(&snake_key, &mut body); + let known_rooms = + services + .sync + .update_snake_sync_request_with_cache(&snake_key, globalsince, &mut body); let mut wake_receiver = services.sync.subscribe_to_wake(sender_user).await; let mut collection = collect_sync_response( @@ -178,7 +184,7 @@ pub(crate) async fn sync_events_v5_route( &body, globalsince, &snake_key, - &collection.known_room_updates, + &collection, ) .await; @@ -391,20 +397,17 @@ async fn commit_sync_collection( body: &sync_events::v5::Request, globalsince: u64, snake_key: &SnakeConnectionsKey, - known_room_updates: &KnownRoomUpdates, + collection: &SyncCollection, ) { - let (.., conn_id) = snake_key; - - if conn_id.is_some() { - for (list_id, rooms) in known_room_updates { - services.sync.update_snake_sync_known_rooms( - snake_key, - list_id.clone(), - rooms.clone(), - globalsince, - ); - } - } + services.sync.store_snake_sync_pending( + snake_key, + collection + .response + .pos + .parse() + .expect("generated sync position"), + collection.known_room_updates.clone(), + ); if body.extensions.to_device.enabled.unwrap_or(false) { services @@ -559,6 +562,22 @@ where active_rooms.push(room_id); } + let mut active_rooms: Vec<_> = active_rooms + .into_iter() + .stream() + .wide_then(|room_id| async move { + let activity = services + .rooms + .timeline + .room_activity(room_id) + .await + .unwrap_or(0); + (room_id, activity) + }) + .collect() + .await; + sort_rooms_by_activity(&mut active_rooms); + let active_rooms: Vec<_> = active_rooms.into_iter().map(at!(0)).collect(); sticky_rooms.extend(active_rooms.iter().map(|room_id| (*room_id).to_owned())); let mut new_known_rooms: BTreeSet = BTreeSet::new(); @@ -626,6 +645,17 @@ fn matches_bool_filter(value: bool, filter: Option) -> bool { filter.is_none_or(|expected| value == expected) } +fn sort_rooms_by_activity(rooms: &mut [(Room, u64)]) +where + Room: Ord, +{ + rooms.sort_unstable_by(|(left_room, left_activity), (right_room, right_activity)| { + right_activity + .cmp(left_activity) + .then_with(|| left_room.cmp(right_room)) + }); +} + /// Rooms we are only invited to have no state locally, so their stripped /// invite state answers for them instead. async fn room_is_encrypted( @@ -1656,6 +1686,15 @@ mod tests { assert!(!matches_bool_filter(true, Some(false))); } + #[test] + fn room_activity_orders_newest_first_with_a_stable_tie_break() { + let mut rooms = [("!b", 3), ("!c", 1), ("!a", 3)]; + + sort_rooms_by_activity(&mut rooms); + + assert_eq!(rooms, [("!a", 3), ("!b", 3), ("!c", 1)]); + } + #[test] fn absent_room_type_filters_match_every_type() { assert!(matches_room_type(&custom_room_type(), &[], &[])); diff --git a/src/database/maps.rs b/src/database/maps.rs index 19e786ebf..457e62764 100644 --- a/src/database/maps.rs +++ b/src/database/maps.rs @@ -193,6 +193,10 @@ pub(super) static MAPS: &[Descriptor] = &[ name: "roomid_joinedcount", ..descriptor::RANDOM_SMALL }, + Descriptor { + name: "roomid_lasttimelinecount", + ..descriptor::RANDOM_SMALL + }, Descriptor { name: "roomid_pduleaves", ..descriptor::RANDOM_SMALL diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 258547615..e56674c2f 100644 --- a/src/service/rooms/timeline/append.rs +++ b/src/service/rooms/timeline/append.rs @@ -405,7 +405,9 @@ impl super::Service { let pdu_id: RawPduId = PduId { shortroomid, shorteventid: count2 }.into(); // Insert pdu - self.db.append_pdu(&pdu_id, pdu, &pdu_json, count2).await; + self.db + .append_pdu(&pdu_id, pdu, room_id, &pdu_json, count2) + .await; drop(insert_lock); diff --git a/src/service/rooms/timeline/data.rs b/src/service/rooms/timeline/data.rs index 69f4f4a67..560bbae84 100644 --- a/src/service/rooms/timeline/data.rs +++ b/src/service/rooms/timeline/data.rs @@ -16,6 +16,7 @@ pub(super) struct Data { eventid_outlierpdu: Arc, eventid_pduid: Arc, pduid_pdu: Arc, + roomid_lasttimelinecount: Arc, userroomid_highlightcount: Arc, userroomid_notificationcount: Arc, pub(super) db: Arc, @@ -35,6 +36,7 @@ impl Data { eventid_outlierpdu: db["eventid_outlierpdu"].clone(), eventid_pduid: db["eventid_pduid"].clone(), pduid_pdu: db["pduid_pdu"].clone(), + roomid_lasttimelinecount: db["roomid_lasttimelinecount"].clone(), userroomid_highlightcount: db["userroomid_highlightcount"].clone(), userroomid_notificationcount: db["userroomid_notificationcount"].clone(), db: args.db.clone(), @@ -59,6 +61,27 @@ impl Data { Ok(last_count) } + pub(super) async fn room_activity(&self, room_id: &RoomId) -> Result { + if let Ok(count) = self + .roomid_lasttimelinecount + .get(room_id) + .await + .deserialized() + { + return Ok(count); + } + + let pdus_rev = self.pdus_rev(room_id, PduCount::max()); + pin_mut!(pdus_rev); + let Some((PduCount::Normal(count), _)) = pdus_rev.try_next().await? else { + return Ok(0); + }; + + self.roomid_lasttimelinecount + .put_raw(room_id.as_bytes(), count.to_be_bytes()); + Ok(count) + } + #[inline] pub(super) async fn latest_pdu_in_room(&self, room_id: &RoomId) -> Result { let pdus_rev = self.pdus_rev(room_id, PduCount::max()); @@ -172,6 +195,7 @@ impl Data { &self, pdu_id: &RawPduId, pdu: &PduEvent, + room_id: &RoomId, json: &CanonicalJsonObject, count: PduCount, ) { @@ -180,6 +204,8 @@ impl Data { self.pduid_pdu.raw_put(pdu_id, Json(json)); self.eventid_pduid.insert(pdu.event_id.as_bytes(), pdu_id); self.eventid_outlierpdu.remove(pdu.event_id.as_bytes()); + self.roomid_lasttimelinecount + .put_raw(room_id.as_bytes(), count.into_unsigned().to_be_bytes()); } pub(super) fn prepend_backfill_pdu( diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs index 8d7ce6d63..715d52855 100644 --- a/src/service/rooms/timeline/mod.rs +++ b/src/service/rooms/timeline/mod.rs @@ -163,6 +163,11 @@ impl Service { self.db.last_timeline_count(room_id).await } + #[tracing::instrument(skip(self), level = "debug")] + pub async fn room_activity(&self, room_id: &RoomId) -> Result { + self.db.room_activity(room_id).await + } + /// Returns the `count` of this pdu's id. pub async fn get_pdu_count(&self, event_id: &EventId) -> Result { self.db.get_pdu_count(event_id).await diff --git a/src/service/sync/mod.rs b/src/service/sync/mod.rs index f3c31433c..6a97377df 100644 --- a/src/service/sync/mod.rs +++ b/src/service/sync/mod.rs @@ -36,9 +36,38 @@ struct SnakeSyncCache { lists: BTreeMap, subscriptions: BTreeMap, known_rooms: BTreeMap>, + acknowledged_pos: u64, + pending: Option, extensions: v5::request::Extensions, } +struct PendingSnakeSync { + pos: u64, + known_rooms: BTreeMap>, +} + +impl SnakeSyncCache { + fn accepts_pos(&self, pos: u64) -> bool { + self.acknowledged_pos == pos + || self + .pending + .as_ref() + .is_some_and(|pending| pending.pos == pos) + } + + fn acknowledge(&mut self, pos: u64) { + if self + .pending + .as_ref() + .is_some_and(|pending| pending.pos == pos) + { + let pending = self.pending.take().expect("pending sync checked above"); + self.known_rooms = pending.known_rooms; + self.acknowledged_pos = pos; + } + } +} + type DbConnections = SyncMutex>; type DbConnectionsKey = (OwnedUserId, OwnedDeviceId, String); pub type SnakeConnectionsKey = (OwnedUserId, OwnedDeviceId, Option); @@ -137,6 +166,13 @@ impl Service { self.snake_connections.lock().contains_key(key) } + pub fn snake_connection_accepts_pos(&self, key: &SnakeConnectionsKey, pos: u64) -> bool { + self.snake_connections + .lock() + .get(key) + .is_some_and(|cache| cache.lock().accepts_pos(pos)) + } + pub fn forget_snake_sync_connection(&self, key: &SnakeConnectionsKey) { self.snake_connections.lock().remove(key); } @@ -144,6 +180,7 @@ impl Service { pub fn update_snake_sync_request_with_cache( &self, snake_key: &SnakeConnectionsKey, + pos: u64, request: &mut v5::Request, ) -> BTreeMap> { let mut cache = self.snake_connections.lock(); @@ -155,6 +192,8 @@ impl Service { let cached = &mut cached.lock(); drop(cache); + cached.acknowledge(pos); + //v5::Request::try_from_http_request(req, path_args); for (list_id, list) in &mut request.lists { if let Some(cached_list) = cached.lists.get(list_id) { @@ -244,14 +283,12 @@ impl Service { cached.known_rooms.clone() } - pub fn update_snake_sync_known_rooms( + pub fn store_snake_sync_pending( &self, key: &SnakeConnectionsKey, - list_id: String, - new_cached_rooms: BTreeSet, - globalsince: u64, + pos: u64, + known_rooms: BTreeMap>, ) { - assert!(key.2.is_some(), "Some(conn_id) required for this call"); let mut cache = self.snake_connections.lock(); let cached = Arc::clone( cache @@ -261,20 +298,17 @@ impl Service { let cached = &mut cached.lock(); drop(cache); - for (room_id, lastsince) in cached - .known_rooms - .entry(list_id.clone()) - .or_default() - .iter_mut() - { - if !new_cached_rooms.contains(room_id) { - *lastsince = 0; + let mut pending_rooms = cached.known_rooms.clone(); + for (list_id, rooms) in known_rooms { + let list = pending_rooms.entry(list_id).or_default(); + for last_seen in list.values_mut() { + *last_seen = 0; + } + for room_id in rooms { + list.insert(room_id, pos); } } - let list = cached.known_rooms.entry(list_id).or_default(); - for room_id in new_cached_rooms { - list.insert(room_id, globalsince); - } + cached.pending = Some(PendingSnakeSync { pos, known_rooms: pending_rooms }); } pub fn update_snake_sync_subscriptions( @@ -386,4 +420,31 @@ mod tests { receiver.changed().await.unwrap(); } + + #[test] + fn pending_rooms_are_committed_only_when_the_response_pos_is_acknowledged() { + let room_id = OwnedRoomId::try_from("!room:example.com").unwrap(); + let mut cache = SnakeSyncCache { + pending: Some(PendingSnakeSync { + pos: 42, + known_rooms: BTreeMap::from([( + "list".to_owned(), + BTreeMap::from([(room_id, 42)]), + )]), + }), + ..Default::default() + }; + + assert!(cache.known_rooms.is_empty()); + assert!(cache.accepts_pos(0)); + assert!(cache.accepts_pos(42)); + assert!(!cache.accepts_pos(41)); + + cache.acknowledge(0); + assert!(cache.known_rooms.is_empty()); + + cache.acknowledge(42); + assert_eq!(cache.acknowledged_pos, 42); + assert_eq!(cache.known_rooms["list"].len(), 1); + } }