fix(sync): Order sliding sync rooms by activity

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);
 

src/service/rooms/timeline/data.rs

@@ -16,6 +16,7 @@ pub(super) struct Data {
 	eventid_outlierpdu: Arc<Map>,
 	eventid_pduid: Arc<Map>,
 	pduid_pdu: Arc<Map>,
+	roomid_lasttimelinecount: Arc<Map>,
 	userroomid_highlightcount: Arc<Map>,
 	userroomid_notificationcount: Arc<Map>,
 	pub(super) db: Arc<Database>,
@@ -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<u64> {
+		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<PduEvent> {
 		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(

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<u64> {
+		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<PduCount> {
 		self.db.get_pdu_count(event_id).await

src/service/sync/mod.rs

@@ -36,9 +36,38 @@ struct SnakeSyncCache {
 	lists: BTreeMap<String, v5::request::List>,
 	subscriptions: BTreeMap<OwnedRoomId, v5::request::RoomSubscription>,
 	known_rooms: BTreeMap<String, BTreeMap<OwnedRoomId, u64>>,
+	acknowledged_pos: u64,
+	pending: Option<PendingSnakeSync>,
 	extensions: v5::request::Extensions,
 }
 
+struct PendingSnakeSync {
+	pos: u64,
+	known_rooms: BTreeMap<String, BTreeMap<OwnedRoomId, u64>>,
+}
+
+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<K, V> = SyncMutex<BTreeMap<K, V>>;
 type DbConnectionsKey = (OwnedUserId, OwnedDeviceId, String);
 pub type SnakeConnectionsKey = (OwnedUserId, OwnedDeviceId, Option<String>);
@@ -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<String, BTreeMap<OwnedRoomId, u64>> {
 		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<OwnedRoomId>,
-		globalsince: u64,
+		pos: u64,
+		known_rooms: BTreeMap<String, BTreeSet<OwnedRoomId>>,
 	) {
-		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);
+	}
 }