diff --git a/hive-matrix-mcp/src/handlers.rs b/hive-matrix-mcp/src/handlers.rs index 15e893b62..0e840f412 100644 --- a/hive-matrix-mcp/src/handlers.rs +++ b/hive-matrix-mcp/src/handlers.rs @@ -1253,10 +1253,21 @@ pub async fn collect_unread(client: &Client) -> Vec #[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, ) -> 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; }; diff --git a/hive-matrix-mcp/src/main.rs b/hive-matrix-mcp/src/main.rs index bad0ad846..51f6fb297 100644 --- a/hive-matrix-mcp/src/main.rs +++ b/hive-matrix-mcp/src/main.rs @@ -781,15 +781,24 @@ fn account_sync_loop(client: &Client, tag: Option) -> 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::)); + // Both loops run this after each response, with the rooms it touched. + let sweep = move |changed: Vec| { 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) -> 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 diff --git a/hive-matrix-mcp/src/sliding.rs b/hive-matrix-mcp/src/sliding.rs index a48407765..a04e07d4d 100644 --- a/hive-matrix-mcp/src/sliding.rs +++ b/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(client: &Client, after_update: F) -> Result<()> where - F: Fn() -> Fut, + F: Fn(Vec) -> Fut, Fut: Future, { 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"); diff --git a/hive-matrix-mcp/src/timeline.rs b/hive-matrix-mcp/src/timeline.rs index 5b53277b1..42846ae4e 100644 --- a/hive-matrix-mcp/src/timeline.rs +++ b/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), +} + +/// 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, now: Instant, changed: Vec) -> 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( + notified: &HashSet, + checked: Option<&HashSet>, + unread: &HashSet, +) -> Vec { + 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( client: &Client, notified: &Mutex>, 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 = ids + .iter() + .filter_map(|id| client.get_room(id)) + .filter(|room| room.state() == matrix_sdk::RoomState::Joined) + .collect(); + let checked: HashSet = + 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 = 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 = { - 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 { + 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 = [room_id!("!a:x"), room_id!("!b:x"), room_id!("!c:x")] + .map(ToOwned::to_owned) + .into(); + let unread: HashSet = [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(¬ified, 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 = [room_id!("!a:x"), room_id!("!c:x")] + .map(ToOwned::to_owned) + .into(); + assert_eq!( + stale_rooms(¬ified, Some(&checked), &unread), + rooms(&["!a:x"]) + ); + } +}