hive-matrix: the unread sweep after a sync looks at the rooms the sync touched, not at every joined room

Each room the sweep looks at costs a live /messages fetch of up to 50 events, and the sweep ran for every joined room after every sync response, so a busy account fetched thousands of times an hour whatever the sync itself carried. The sweep now covers the rooms the response names (sliding sync's UpdateSummary rooms, classic's joined rooms), with a full pass first and every five minutes.

hive-matrix-mcp/src/handlers.rs

@@ -1253,10 +1253,21 @@ pub async fn collect_unread(client: &Client) -> Vec<crate::protocol::RoomUnread>
 #[must_use]
 pub async fn collect_unread_with_ids(
     client: &Client,
+) -> Vec<(matrix_sdk::ruma::OwnedRoomId, crate::protocol::RoomUnread)> {
+    collect_unread_in(client, client.joined_rooms()).await
+}
+
+/// [`collect_unread_with_ids`] over just `rooms`. Each room costs one live
+/// `/messages` fetch (see `recent_events`), so the post-sync sweep passes
+/// only the rooms the sync touched.
+#[must_use]
+pub async fn collect_unread_in(
+    client: &Client,
+    rooms: Vec<matrix_sdk::Room>,
 ) -> Vec<(matrix_sdk::ruma::OwnedRoomId, crate::protocol::RoomUnread)> {
     use crate::protocol::RoomUnread;
     let mut result = Vec::new();
-    for room in client.joined_rooms() {
+    for room in rooms {
         let Some((_, sender, body)) = room_unread_state(client, &room).await else {
             continue;
         };

hive-matrix-mcp/src/main.rs

@@ -781,15 +781,24 @@ fn account_sync_loop(client: &Client, tag: Option<String>) -> SyncLoop {
     // each sweep (see the sweep fns) so re-invites / new messages re-push.
     let invite_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new()));
     let unread_notified = Arc::new(tokio::sync::Mutex::new(std::collections::HashSet::new()));
-    // Both loops run this after each response.
-    let sweep = move || {
+    // When the sweep last covered every room, so the others are only looked
+    // at again after `timeline::FULL_SWEEP_EVERY` (see `timeline::plan`).
+    let last_full = Arc::new(tokio::sync::Mutex::new(None::<std::time::Instant>));
+    // Both loops run this after each response, with the rooms it touched.
+    let sweep = move |changed: Vec<matrix_sdk::ruma::OwnedRoomId>| {
         let client = cb_client.clone();
         let invite_notified = invite_notified.clone();
         let unread_notified = unread_notified.clone();
+        let last_full = last_full.clone();
         let tag = tag.clone();
         async move {
             timeline::sweep_invites(&client, &invite_notified, tag.as_deref()).await;
-            timeline::sweep_unread(&client, &unread_notified, tag.as_deref()).await;
+            let now = std::time::Instant::now();
+            let scope = timeline::plan(*last_full.lock().await, now, changed);
+            if scope == timeline::Scope::All {
+                *last_full.lock().await = Some(now);
+            }
+            timeline::sweep_unread(&client, &unread_notified, tag.as_deref(), scope).await;
         }
     };
     Box::pin(async move {
@@ -803,8 +812,8 @@ fn account_sync_loop(client: &Client, tag: Option<String>) -> SyncLoop {
             tracing::info!("syncing with classic /sync");
         }
         sync_client
-            .sync_with_callback(SyncSettings::default().filter(sync_filter()), |_response| {
-                let sweep = sweep();
+            .sync_with_callback(SyncSettings::default().filter(sync_filter()), |response| {
+                let sweep = sweep(response.rooms.joined.keys().cloned().collect());
                 async move {
                     sweep.await;
                     matrix_sdk::LoopCtrl::Continue

hive-matrix-mcp/src/sliding.rs

@@ -15,6 +15,7 @@ use std::path::Path;
 
 use anyhow::{Result, anyhow, bail};
 use futures_util::StreamExt as _;
+use matrix_sdk::ruma::OwnedRoomId;
 use matrix_sdk::ruma::events::StateEventType;
 use matrix_sdk::sliding_sync::{SlidingSyncList, SlidingSyncMode, Version};
 use matrix_sdk::{Client, SqliteCryptoStore};
@@ -115,12 +116,13 @@ fn required_state() -> Vec<(StateEventType, String)> {
     .collect()
 }
 
-/// Poll sliding sync for `client`, calling `after_update` once per response.
+/// Poll sliding sync for `client`, calling `after_update` once per response with
+/// the rooms that response touched.
 /// Returns `Err` once 5 polls in a row have failed, so the caller can fall
 /// back to `/sync`.
 pub async fn run<F, Fut>(client: &Client, after_update: F) -> Result<()>
 where
-    F: Fn() -> Fut,
+    F: Fn(Vec<OwnedRoomId>) -> Fut,
     Fut: Future<Output = ()>,
 {
     let list = SlidingSyncList::builder("all-rooms")
@@ -143,9 +145,9 @@ where
     let mut errors = ErrorRun::default();
     while let Some(update) = stream.next().await {
         match update {
-            Ok(_) => {
+            Ok(summary) => {
                 errors.ok();
-                after_update().await;
+                after_update(summary.rooms).await;
             }
             Err(e) => {
                 tracing::warn!(error = %e, "sliding sync poll failed");

hive-matrix-mcp/src/timeline.rs

@@ -16,6 +16,7 @@
 
 use std::collections::HashSet;
 use std::hash::BuildHasher;
+use std::time::{Duration, Instant};
 
 use matrix_sdk::{
     Client,
@@ -25,6 +26,48 @@ use tokio::sync::Mutex;
 
 use crate::{handlers, wake};
 
+/// How long the rooms the sync did not touch go unchecked before a sweep
+/// covers all of them again. The slow safety net behind the per-response
+/// sweep of the changed rooms: a room the sync failed to name still gets
+/// its unread state looked at within this time. Each room checked costs a
+/// live `/messages` fetch, so this is also what the quiet rooms cost.
+const FULL_SWEEP_EVERY: Duration = Duration::from_mins(5);
+
+/// Which joined rooms a sweep looks at.
+#[derive(Debug, PartialEq, Eq)]
+pub enum Scope {
+    /// Every joined room.
+    All,
+    /// The rooms the last sync response touched, and no others.
+    Changed(Vec<OwnedRoomId>),
+}
+
+/// The scope of the sweep after a sync response that touched `changed`:
+/// everything on the first sweep and again every five minutes, the changed
+/// rooms otherwise.
+#[must_use]
+pub fn plan(last_full: Option<Instant>, now: Instant, changed: Vec<OwnedRoomId>) -> Scope {
+    match last_full {
+        Some(at) if now.saturating_duration_since(at) < FULL_SWEEP_EVERY => Scope::Changed(changed),
+        _ => Scope::All,
+    }
+}
+
+/// The rooms holding a todo that this sweep has looked at and found read:
+/// `notified` less `unread`, restricted to `checked` when the sweep did not
+/// look at every room (a room it skipped keeps its todo).
+fn stale_rooms<S: BuildHasher>(
+    notified: &HashSet<OwnedRoomId, S>,
+    checked: Option<&HashSet<OwnedRoomId>>,
+    unread: &HashSet<OwnedRoomId>,
+) -> Vec<OwnedRoomId> {
+    notified
+        .iter()
+        .filter(|id| !unread.contains(*id) && checked.is_none_or(|c| c.contains(*id)))
+        .cloned()
+        .collect()
+}
+
 /// Push a *todo* (loose-ends v2) for each joined room carrying
 /// unread notifications, and clear the todo for rooms that have been read.
 /// Replaces the old direct-wake path: instead of firing an all-rooms wake,
@@ -47,21 +90,32 @@ pub async fn sweep_unread<S: BuildHasher>(
     client: &Client,
     notified: &Mutex<HashSet<OwnedRoomId, S>>,
     account_tag: Option<&str>,
+    scope: Scope,
 ) {
-    let unread = handlers::collect_unread_with_ids(client).await;
+    let (rooms, checked) = match scope {
+        Scope::All => (client.joined_rooms(), None),
+        Scope::Changed(ids) => {
+            let rooms: Vec<matrix_sdk::Room> = ids
+                .iter()
+                .filter_map(|id| client.get_room(id))
+                .filter(|room| room.state() == matrix_sdk::RoomState::Joined)
+                .collect();
+            let checked: HashSet<OwnedRoomId> =
+                rooms.iter().map(|room| room.room_id().to_owned()).collect();
+            (rooms, Some(checked))
+        }
+    };
+    if rooms.is_empty() && checked.is_some() {
+        // A response that touched no joined room: nothing to look at.
+        return;
+    }
+    let unread = handlers::collect_unread_in(client, rooms).await;
     let unread_ids: HashSet<OwnedRoomId> = unread.iter().map(|(id, _)| id.clone()).collect();
 
     // Rooms we previously pushed a todo for that are no longer unread → the
     // agent read them; clear their todo. Snapshot under the lock, send
     // outside it, drop from `notified` on success (retry next tick on fail).
-    let stale: Vec<OwnedRoomId> = {
-        let active = notified.lock().await;
-        active
-            .iter()
-            .filter(|id| !unread_ids.contains(*id))
-            .cloned()
-            .collect()
-    };
+    let stale = stale_rooms(&*notified.lock().await, checked.as_ref(), &unread_ids);
     for id in stale {
         if wake::send_todo_clear(id.as_str()).await.is_ok() {
             notified.lock().await.remove(&id);
@@ -258,3 +312,70 @@ mod invite_tests {
         assert!(!is_local(None, Some(user_id!("@x:catgirl.cloud"))));
     }
 }
+
+#[cfg(test)]
+mod sweep_scope_tests {
+    use std::collections::HashSet;
+    use std::time::{Duration, Instant};
+
+    use matrix_sdk::ruma::{OwnedRoomId, room_id};
+
+    use super::{FULL_SWEEP_EVERY, Scope, plan, stale_rooms};
+
+    fn rooms(ids: &[&str]) -> Vec<OwnedRoomId> {
+        ids.iter()
+            .map(|id| OwnedRoomId::try_from(*id).expect("room id"))
+            .collect()
+    }
+
+    #[test]
+    fn the_first_sweep_covers_every_room() {
+        assert_eq!(plan(None, Instant::now(), rooms(&["!a:x"])), Scope::All);
+    }
+
+    #[test]
+    fn between_full_sweeps_only_the_rooms_the_sync_touched_are_looked_at() {
+        let now = Instant::now();
+        let last = now
+            .checked_sub(Duration::from_secs(10))
+            .expect("an earlier instant");
+        assert_eq!(
+            plan(Some(last), now, rooms(&["!a:x"])),
+            Scope::Changed(rooms(&["!a:x"]))
+        );
+        // A response that touched nothing looks at nothing.
+        assert_eq!(
+            plan(Some(last), now, Vec::new()),
+            Scope::Changed(Vec::new())
+        );
+    }
+
+    #[test]
+    fn every_so_often_the_sweep_covers_every_room_again() {
+        let now = Instant::now();
+        let last = now
+            .checked_sub(FULL_SWEEP_EVERY)
+            .expect("an earlier instant");
+        assert_eq!(plan(Some(last), now, rooms(&["!a:x"])), Scope::All);
+    }
+
+    #[test]
+    fn a_room_the_sweep_skipped_keeps_its_todo_and_a_read_one_loses_it() {
+        let notified: HashSet<OwnedRoomId> = [room_id!("!a:x"), room_id!("!b:x"), room_id!("!c:x")]
+            .map(ToOwned::to_owned)
+            .into();
+        let unread: HashSet<OwnedRoomId> = [room_id!("!c:x").to_owned()].into();
+        // A full sweep: !a and !b are read now, !c is still unread.
+        let mut all = stale_rooms(&notified, None, &unread);
+        all.sort();
+        assert_eq!(all, rooms(&["!a:x", "!b:x"]));
+        // A sweep of !a and !c only: !b was not looked at, so its todo stays.
+        let checked: HashSet<OwnedRoomId> = [room_id!("!a:x"), room_id!("!c:x")]
+            .map(ToOwned::to_owned)
+            .into();
+        assert_eq!(
+            stale_rooms(&notified, Some(&checked), &unread),
+            rooms(&["!a:x"])
+        );
+    }
+}