macltok

public

Star0

Fork 0

Watch0

89 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `21b9227729a4`

tarball

atagen <boss@atagen.co>2026-10-25 13:25 UTC

12 files changed+766-318

threads: count from the store and stop list counts drifting

Thread unreads counted over the event cache's memory window moved whenever a room closed, the cache shrank or the app restarted, and recounting made an event cache for every cold room. They are now counted from the event cache store's thread relations and kept in the state store, so rooms() never touches the event cache. Replays still inflated thread list counts, since an unknown latest timestamp read as zero, so only a reply sync just inserted, newer than a dated latest and not seen before now counts. A lag refetches the first page behind a ThreadListReset, and a leave, kick or ban closes the list with a ThreadListClosure.

Modified file: docs/design-decisions.md+41 additions−22 deletions@@ -138,16 +138,23 @@ can
 the sdk's `ThreadListService` was passed over. it never adds a thread that wasn't
 on a page already loaded, never reorders, and adds one to the count for every
 event the cache hands it, replays and back pagination included. here a reply
-counts once, only when it's newer than the thread's latest, and a reply sync
-brings to a thread we haven't loaded fetches its root and adds it. one a back
-pagination brings doesn't, since that says nothing about the thread being
-active. a redacted reply doesn't uncount, an edit doesn't change the latest
-reply's preview, and a preview's own reply target comes back `Unavailable`
-rather than fetched. a lagged listener misses whatever it skipped until the
-thread moves again
-
-a list's media belongs to its own scope, bounded like an event context's so a
-long open list sheds the previews its threads moved past, and released on close
+counts once, by id, when sync just brought it and it's newer than a latest
+whose time we know, by the sdk's corrected timestamp. the reply a thread already
+ends on only dates it, since a summary from the event cache names its latest
+reply without the event. replays, reloads from the store and back pagination
+never count. a reply sync brings to a thread the list hasn't got fetches its
+root, four at a time, and adds it. one from anywhere else doesn't, since that
+says nothing about the thread being active, and a root the server doesn't have
+isn't asked for again. a redacted reply doesn't uncount, an edit doesn't change
+the latest reply's preview, and a preview's own reply target comes back
+`Unavailable` rather than fetched
+
+a lagged listener can't know what it missed, so the list fetches its first page
+again and sends a `ThreadListReset` with it. leaving the room, or being kicked
+or banned, closes the list with `ThreadListClosed` naming why, the same as
+`Close` does. a list's media belongs to its own scope, bounded like an event
+context's so a long open list sheds the previews its threads moved past, and
+released on close
 
 ## warming up to send
 
@@ -354,23 +361,35 @@ sdk ui types stay behind the boundary
 
 `rooms` normalises on demand with bounded concurrency and sorts by latest event,
 then name, then id. `RoomsChanged` is an invalidation, coalesced so one sync
-touching fifty rooms is one event. a left room stays in the list with its
+touching fifty rooms is one event. a sync's room info lands before the event
+cache has stored its events, so the watcher waits a tenth of a second after
+anything wakes it and takes the lot as one. an event cache slower than that
+costs a second one. a left room stays in the list with its
 membership until it's forgotten. push rule changes never touch room info, so
 `NotificationSettingsChanged` brings a `RoomsChanged` along for the modes in
 the summaries
 
 with threads on, the sdk leaves thread events out of a room's counts and keeps
-none of its own for them. `thread_unreads` works them out per root the way the
-sdk does a room's, over the events its cache holds in memory, newest first until
-a receipt for that thread (msc3771), an unthreaded one, or the user's own reply.
-a main-timeline receipt doesn't reach into threads. a thread's mentions also count
-in the room's `unread_mentions`, so nobody misses one, while its messages and
-notifications stay apart, since they don't move the main timeline's read marker.
-thread events and threaded receipts leave room info alone, so the room list
-watcher also recounts a room on each event cache update and each receipt, and
-only says `RoomsChanged` when the counts moved. counting in the sdk would be
-cheaper and survive memory being trimmed, but that's a behaviour change in the
-fork
+none of its own for them. `thread_unreads` counts them per root from the event
+cache store, which keeps every reply it ever took whatever it holds in memory,
+so closing a room, the cache trimming itself or a restart leaves them be. a
+reply counts once it's newer than a receipt for its thread (msc3771), an
+unthreaded one, or the user's own reply. a main-timeline receipt doesn't reach
+into threads. the store has lost stream order between a thread and the rest of
+the room, so an event's timestamp stands for its place, and a receipt on an
+event the store never had falls back to the receipt's own time. a thread's
+mentions also count in the room's `unread_mentions`, so nobody misses one,
+while its messages and notifications stay apart, since they don't move the main
+timeline's read marker
+
+the counts are kept in the state store per room, only for threads with
+something unread, so `rooms` reads them without touching the event cache and a
+cold room never gets one made for it. thread events and threaded receipts leave
+room info alone, so the room list watcher recounts on its own. a synced thread
+event names the thread to count, an event cache update recounts the room's
+threads already holding something unread, since a late decryption can turn up
+a mention, and so does a receipt of the user's own. counting in the sdk would
+be cheaper, but that's a behaviour change in the fork
 
 ## link previews
 

Modified file: src/engine/context.rs+4 additions−13 deletions@@ -14,11 +14,11 @@ use tokio::sync::{Mutex, broadcast::error::RecvError};
 
 use crate::{
     Error, ErrorKind, Event, EventContext, EventContextCloseReason, EventContextCommand,
-    EventContextPage, EventContextSession, MediaId, Sender, TimelineEvent, TimelineItem,
+    EventContextPage, EventContextSession, Sender, TimelineEvent, TimelineItem,
     engine::MediaService,
     normalise::{
-        MediaRegistration, MediaScope, apply_profiles, normalise_context, normalise_observed,
-        resolve_replies,
+        MediaRegistration, MediaScope, apply_profiles, member_sender, normalise_context,
+        normalise_observed, resolve_replies,
     },
     transport::EventSender,
 };
@@ -525,16 +525,7 @@ async fn profiles<'a>(
     unique.sort();
     unique.dedup();
     join_all(unique.into_iter().map(|user_id| async move {
-        let member = room.get_member(user_id).await.ok().flatten()?;
-        Some((
-            user_id.clone(),
-            Sender {
-                user_id: user_id.clone(),
-                display_name: member.display_name().map(str::to_owned),
-                avatar: member.avatar_url().map(MediaId::plain),
-                name_ambiguous: member.name_ambiguous(),
-            },
-        ))
+        Some((user_id.clone(), member_sender(room, user_id).await?))
     }))
     .await
     .into_iter()

Modified file: src/engine/thread_list.rs+408 additions−152 deletions@@ -1,10 +1,11 @@
 use std::{
-    collections::{HashMap, hash_map::Entry},
+    collections::{HashMap, HashSet, VecDeque},
     sync::Arc,
 };
 
+use imbl::Vector;
 use matrix_sdk::{
-    Client, Room,
+    Client, Room, RoomState,
     deserialized_responses::TimelineEvent as SdkTimelineEvent,
     event_cache::{EventsOrigin, RoomEventCacheSubscriber, RoomEventCacheUpdate},
     room::ListThreadsOptions,
@@ -17,16 +18,21 @@ use tokio::sync::{broadcast::error::RecvError, mpsc};
 use super::{
     MediaService,
     commands::Job,
-    executor::{TaskSet, is_panic, resume_unwind},
+    executor::{AbortHandle, TaskSet, is_panic, resume_unwind},
 };
 use crate::{
-    Error, ErrorKind, Event, HistoryState, MediaId, Sender, ThreadListCommand, ThreadListEntry,
+    Error, ErrorKind, Event, HistoryState, ThreadListClosure, ThreadListCommand, ThreadListEntry,
     ThreadSummary, TimelineItem,
-    normalise::{MediaRegistration, MediaScope, normalise_observed, resolve_replies},
+    normalise::{
+        MediaRegistration, MediaScope, member_sender, normalise_observed, resolve_replies,
+    },
     transport::{EventSender, Responder},
 };
 
 const MAX_PAGE: u16 = 50;
+/// the first page again after a lag, when nobody has paginated yet
+const DEFAULT_PAGE: u16 = 20;
+const MAX_ROOT_FETCHES: usize = 4;
 
 struct Normalised {
     event_id: OwnedEventId,
@@ -52,7 +58,16 @@ enum Arrival {
         room_id: OwnedRoomId,
         generation: u64,
         root_event_id: OwnedEventId,
-        built: Option<Built>,
+        fetched: Fetched,
+    },
+    Lagged {
+        room_id: OwnedRoomId,
+        generation: u64,
+    },
+    Left {
+        room_id: OwnedRoomId,
+        generation: u64,
+        reason: ThreadListClosure,
     },
 }
 
@@ -60,12 +75,20 @@ enum Live {
     Reply {
         root_event_id: OwnedEventId,
         reply: Normalised,
-        /// back pagination, which says nothing about a thread being active
-        paginated: bool,
+        /// sync brought it into the room just now. a replay, a reload from the store or a
+        /// back pagination says nothing about the thread moving
+        fresh: bool,
     },
     Root(Normalised),
 }
 
+enum Fetched {
+    Built(Built),
+    /// the server doesn't have it, or it isn't a thread root we can show
+    Missing,
+    Failed,
+}
+
 struct Page {
     entries: Vec<Built>,
     token: Option<String>,
@@ -76,18 +99,32 @@ struct Built {
     media: Vec<MediaRegistration>,
 }
 
+/// what a list is loading, and who waits on it
+enum Load {
+    Idle,
+    Page(Vec<Responder>),
+    /// the first page again, replacing everything. whoever waits wants the page after it
+    Refresh(Vec<Responder>),
+}
+
 struct List {
     room: Room,
     generation: u64,
     scope: MediaScope,
-    entries: Vec<ThreadListEntry>,
+    entries: Vector<ThreadListEntry>,
     token: Option<String>,
     end_reached: bool,
-    /// non-empty while a page is in flight, everyone waiting on it
-    paginating: Vec<Responder>,
-    /// roots being fetched, with the replies that arrived meanwhile
+    page_limit: u16,
+    load: Load,
+    refresh_owed: bool,
+    /// replies already counted, so the same event synced twice counts once
+    counted: HashSet<OwnedEventId>,
+    /// roots being fetched or waiting for a slot, with the replies that arrived meanwhile
     fetching: HashMap<OwnedEventId, Vec<Normalised>>,
-    listener: super::executor::AbortHandle,
+    waiting_roots: VecDeque<OwnedEventId>,
+    fetches_in_flight: usize,
+    unfetchable: HashSet<OwnedEventId>,
+    listener: AbortHandle,
 }
 
 pub(super) async fn run(
@@ -142,7 +179,15 @@ impl ThreadLists {
             ThreadListCommand::Paginate { room_id, limit } => {
                 self.paginate(room_id, limit, reply).await;
             }
-            ThreadListCommand::Close(room_id) => self.close(room_id, reply).await,
+            ThreadListCommand::Close(room_id) => {
+                let Some(list) = self.lists.remove(&room_id) else {
+                    return reply.done(Err(not_open()));
+                };
+                let closed = self
+                    .finish(room_id, list, ThreadListClosure::Requested)
+                    .await;
+                reply.done(closed);
+            }
         }
     }
 
@@ -150,7 +195,7 @@ impl ThreadLists {
         if let Some(list) = self.lists.get(&room_id) {
             let reset = Event::ThreadListReset {
                 room_id,
-                entries: list.entries.iter().cloned().collect(),
+                entries: list.entries.clone(),
             };
             return reply.done(self.emit(reset).await);
         }
@@ -171,17 +216,23 @@ impl ThreadLists {
                 room,
                 generation,
                 scope: self.media.context_scope(),
-                entries: Vec::new(),
+                entries: Vector::new(),
                 token: None,
                 end_reached: false,
-                paginating: Vec::new(),
+                page_limit: DEFAULT_PAGE,
+                load: Load::Idle,
+                refresh_owed: false,
+                counted: HashSet::new(),
                 fetching: HashMap::new(),
+                waiting_roots: VecDeque::new(),
+                fetches_in_flight: 0,
+                unfetchable: HashSet::new(),
                 listener,
             },
         );
         let reset = Event::ThreadListReset {
             room_id,
-            entries: imbl::Vector::new(),
+            entries: Vector::new(),
         };
         reply.done(self.emit(reset).await);
     }
@@ -196,17 +247,26 @@ impl ThreadLists {
         let Some(list) = self.lists.get_mut(&room_id) else {
             return reply.done(Err(not_open()));
         };
+        list.page_limit = limit;
+        if let Load::Page(waiting) | Load::Refresh(waiting) = &mut list.load {
+            return waiting.push(reply);
+        }
         if list.end_reached {
             return reply.done(Ok(()));
         }
-        list.paginating.push(reply);
-        if list.paginating.len() > 1 {
+        list.load = Load::Page(vec![reply]);
+        self.load(&room_id, false).await;
+    }
+
+    /// starts the fetch `list.load` already names
+    async fn load(&mut self, room_id: &OwnedRoomId, refresh: bool) {
+        let Some(list) = self.lists.get(room_id) else {
             return;
-        }
+        };
         let fetch = fetch_page(
             list.room.clone(),
-            list.token.clone(),
-            limit,
+            if refresh { None } else { list.token.clone() },
+            list.page_limit,
             self.current_user_id.clone(),
         );
         let generation = list.generation;
@@ -222,22 +282,27 @@ impl ThreadLists {
         });
         let _ = self
             .emit(Event::ThreadListHistoryChanged {
-                room_id,
+                room_id: room_id.clone(),
                 state: HistoryState::Loading,
             })
             .await;
     }
 
-    async fn close(&mut self, room_id: OwnedRoomId, reply: Responder) {
-        let Some(list) = self.lists.remove(&room_id) else {
-            return reply.done(Err(not_open()));
-        };
+    /// the list is already out of `lists`
+    async fn finish(
+        &mut self,
+        room_id: OwnedRoomId,
+        list: List,
+        reason: ThreadListClosure,
+    ) -> Result<(), Error> {
         list.listener.abort();
-        for waiting in list.paginating {
-            waiting.done(Err(Error::interrupted()));
+        if let Load::Page(waiting) | Load::Refresh(waiting) = list.load {
+            for reply in waiting {
+                reply.done(Err(Error::interrupted()));
+            }
         }
         self.media.release(&list.scope).await;
-        reply.done(self.emit(Event::ThreadListClosed { room_id }).await);
+        self.emit(Event::ThreadListClosed { room_id, reason }).await
     }
 
     async fn arrive(&mut self, arrival: Arrival) {
@@ -256,22 +321,36 @@ impl ThreadLists {
                 room_id,
                 generation,
                 root_event_id,
-                built,
+                fetched,
+            } => {
+                self.root_arrived(room_id, generation, root_event_id, fetched)
+                    .await;
+            }
+            Arrival::Lagged {
+                room_id,
+                generation,
             } => {
                 let Some(list) = current(&mut self.lists, &room_id, generation) else {
                     return;
                 };
-                let replies = list.fetching.remove(&root_event_id).unwrap_or_default();
-                let Some(mut built) = built else {
-                    return;
-                };
-                if list.position(&root_event_id).is_some() {
+                if matches!(list.load, Load::Idle) {
+                    list.load = Load::Refresh(Vec::new());
+                    self.load(&room_id, true).await;
+                } else {
+                    list.refresh_owed = true;
+                }
+            }
+            Arrival::Left {
+                room_id,
+                generation,
+                reason,
+            } => {
+                if current(&mut self.lists, &room_id, generation).is_none() {
                     return;
                 }
-                for reply in replies {
-                    built.media.extend(add_reply(&mut built.entry, reply));
+                if let Some(list) = self.lists.remove(&room_id) {
+                    let _ = self.finish(room_id, list, reason).await;
                 }
-                self.publish(&room_id, vec![built]).await;
             }
         }
     }
@@ -285,28 +364,36 @@ impl ThreadLists {
         let Some(list) = current(&mut self.lists, &room_id, generation) else {
             return;
         };
-        let waiting = std::mem::take(&mut list.paginating);
+        let (waiting, refresh) = match std::mem::replace(&mut list.load, Load::Idle) {
+            Load::Page(waiting) => (waiting, false),
+            Load::Refresh(waiting) => (waiting, true),
+            Load::Idle => return,
+        };
         let (result, state) = match page {
             Ok(page) => {
                 list.end_reached = page.token.is_none();
                 list.token = page.token;
-                let fresh = page
-                    .entries
-                    .into_iter()
-                    .filter(|built| {
-                        list.position(&built.entry.root_event_id)
-                            .is_none_or(|index| {
-                                list.entries[index].latest_activity_ms
-                                    <= built.entry.latest_activity_ms
-                            })
-                    })
-                    .collect();
                 let state = if list.end_reached {
                     HistoryState::StartReached
                 } else {
                     HistoryState::Idle
                 };
-                self.publish(&room_id, fresh).await;
+                if refresh {
+                    self.replace(&room_id, page.entries).await;
+                } else {
+                    let fresh = page
+                        .entries
+                        .into_iter()
+                        .filter(|built| {
+                            list.position(&built.entry.root_event_id)
+                                .is_none_or(|index| {
+                                    list.entries[index].latest_activity_ms
+                                        <= built.entry.latest_activity_ms
+                                })
+                        })
+                        .collect();
+                    self.publish(&room_id, fresh).await;
+                }
                 (Ok(()), state)
             }
             Err(error) => {
@@ -315,10 +402,32 @@ impl ThreadLists {
             }
         };
         let _ = self
-            .emit(Event::ThreadListHistoryChanged { room_id, state })
+            .emit(Event::ThreadListHistoryChanged {
+                room_id: room_id.clone(),
+                state,
+            })
             .await;
-        for reply in waiting {
-            reply.done(result.clone());
+        let carried = if refresh && result.is_ok() {
+            waiting
+        } else {
+            for reply in waiting {
+                reply.done(result.clone());
+            }
+            Vec::new()
+        };
+        let Some(list) = self.lists.get_mut(&room_id) else {
+            return;
+        };
+        if std::mem::take(&mut list.refresh_owed) {
+            list.load = Load::Refresh(carried);
+            self.load(&room_id, true).await;
+        } else if list.end_reached {
+            for reply in carried {
+                reply.done(Ok(()));
+            }
+        } else if !carried.is_empty() {
+            list.load = Load::Page(carried);
+            self.load(&room_id, false).await;
         }
     }
 
@@ -330,38 +439,25 @@ impl ThreadLists {
             Live::Reply {
                 root_event_id,
                 reply,
-                paginated,
+                fresh,
             } => {
                 if let Some(index) = list.position(&root_event_id) {
                     let mut entry = list.entries[index].clone();
-                    let media = add_reply(&mut entry, reply);
+                    let media = add_reply(&mut entry, reply, fresh, &mut list.counted);
                     if entry != list.entries[index] {
                         self.publish(&room_id, vec![Built { entry, media }]).await;
                     }
                     return;
                 }
-                if paginated {
-                    return;
+                if let Some(waiting) = list.fetching.get_mut(&root_event_id) {
+                    return waiting.push(reply);
                 }
-                match list.fetching.entry(root_event_id.clone()) {
-                    Entry::Occupied(mut waiting) => waiting.get_mut().push(reply),
-                    Entry::Vacant(vacant) => {
-                        vacant.insert(vec![reply]);
-                        let room = list.room.clone();
-                        let current_user_id = self.current_user_id.clone();
-                        let arrivals = self.arrivals.clone();
-                        self.tasks.spawn(async move {
-                            let built =
-                                fetch_root(&room, &root_event_id, current_user_id).await;
-                            let _ = arrivals.send(Arrival::Root {
-                                room_id: room.room_id().to_owned(),
-                                generation,
-                                root_event_id,
-                                built,
-                            });
-                        });
-                    }
+                if !fresh || list.unfetchable.contains(&root_event_id) {
+                    return;
                 }
+                list.fetching.insert(root_event_id.clone(), vec![reply]);
+                list.waiting_roots.push_back(root_event_id);
+                self.fetch_roots(&room_id);
             }
             Live::Root(root) => {
                 let Some(index) = list.position(&root.event_id) else {
@@ -377,6 +473,76 @@ impl ThreadLists {
         }
     }
 
+    /// as many waiting roots as there are free slots
+    fn fetch_roots(&mut self, room_id: &OwnedRoomId) {
+        let Some(list) = self.lists.get_mut(room_id) else {
+            return;
+        };
+        while list.fetches_in_flight < MAX_ROOT_FETCHES {
+            let Some(root_event_id) = list.waiting_roots.pop_front() else {
+                return;
+            };
+            list.fetches_in_flight += 1;
+            let room = list.room.clone();
+            let generation = list.generation;
+            let current_user_id = self.current_user_id.clone();
+            let arrivals = self.arrivals.clone();
+            self.tasks.spawn(async move {
+                let fetched = fetch_root(&room, &root_event_id, current_user_id).await;
+                let _ = arrivals.send(Arrival::Root {
+                    room_id: room.room_id().to_owned(),
+                    generation,
+                    root_event_id,
+                    fetched,
+                });
+            });
+        }
+    }
+
+    async fn root_arrived(
+        &mut self,
+        room_id: OwnedRoomId,
+        generation: u64,
+        root_event_id: OwnedEventId,
+        fetched: Fetched,
+    ) {
+        let Some(list) = current(&mut self.lists, &room_id, generation) else {
+            return;
+        };
+        list.fetches_in_flight = list.fetches_in_flight.saturating_sub(1);
+        let replies = list.fetching.remove(&root_event_id).unwrap_or_default();
+        let built = match (fetched, list.position(&root_event_id)) {
+            (_, Some(index)) => {
+                let entry = list.entries[index].clone();
+                let mut built = Built {
+                    entry: entry.clone(),
+                    media: Vec::new(),
+                };
+                for reply in replies {
+                    let media = add_reply(&mut built.entry, reply, true, &mut list.counted);
+                    built.media.extend(media);
+                }
+                (built.entry != entry).then_some(built)
+            }
+            (Fetched::Built(mut built), None) => {
+                for reply in replies {
+                    let media = add_reply(&mut built.entry, reply, true, &mut list.counted);
+                    built.media.extend(media);
+                }
+                Some(built)
+            }
+            (Fetched::Missing, None) => {
+                list.unfetchable.insert(root_event_id);
+                None
+            }
+            (Fetched::Failed, None) => None,
+        };
+        self.fetch_roots(&room_id);
+        if let Some(built) = built {
+            self.publish(&room_id, vec![built]).await;
+        }
+    }
+
     /// the one place a thread list's media is registered. its scope is bounded like an event
     /// context's, so previews a long open list replaced age out, and it goes on close
     async fn publish(&mut self, room_id: &OwnedRoomId, built: Vec<Built>) {
@@ -387,7 +553,6 @@ impl ThreadLists {
             return;
         };
         let mut media = Vec::new();
-        let scope = list.scope.clone();
         let mut upserts = Vec::with_capacity(built.len());
         for Built {
             entry,
@@ -395,9 +560,10 @@ impl ThreadLists {
         } in built
         {
             media.extend(registrations);
-            list.upsert(entry.clone());
+            ThreadListEntry::upsert(&mut list.entries, entry.clone());
             upserts.push(entry);
         }
+        let scope = list.scope.clone();
         self.media.register(&scope, media).await;
         let _ = self
             .emit(Event::ThreadListChanged {
@@ -407,6 +573,32 @@ impl ThreadLists {
             .await;
     }
 
+    /// a refreshed first page, replacing whatever the list held
+    async fn replace(&mut self, room_id: &OwnedRoomId, built: Vec<Built>) {
+        let Some(list) = self.lists.get_mut(room_id) else {
+            return;
+        };
+        let mut media = Vec::new();
+        list.entries = Vector::new();
+        for Built {
+            entry,
+            media: registrations,
+        } in built
+        {
+            media.extend(registrations);
+            ThreadListEntry::upsert(&mut list.entries, entry);
+        }
+        let entries = list.entries.clone();
+        let scope = list.scope.clone();
+        self.media.register(&scope, media).await;
+        let _ = self
+            .emit(Event::ThreadListReset {
+                room_id: room_id.clone(),
+                entries,
+            })
+            .await;
+    }
+
     async fn emit(&self, event: Event) -> Result<(), Error> {
         self.events
             .send(event)
@@ -421,16 +613,6 @@ impl List {
             .iter()
             .position(|entry| &entry.root_event_id == root_event_id)
     }
-
-    fn upsert(&mut self, entry: ThreadListEntry) {
-        if let Some(index) = self.position(&entry.root_event_id) {
-            self.entries.remove(index);
-        }
-        let index = self
-            .entries
-            .partition_point(|known| known.sorts_before(&entry));
-        self.entries.insert(index, entry);
-    }
 }
 
 fn current<'a>(
@@ -447,24 +629,40 @@ fn not_open() -> Error {
     Error::new(ErrorKind::TimelineNotOpen, "thread list is not open")
 }
 
-/// a reply newer than the thread's latest counts once. the event cache replays what it
-/// already had on a reset or a reload, and back pagination brings older ones, neither of
-/// which the server's count left out
-fn add_reply(entry: &mut ThreadListEntry, reply: Normalised) -> Vec<MediaRegistration> {
-    if entry.summary.latest_event_id.as_ref() == Some(&reply.event_id) {
-        entry.latest_reply = Some(reply.item);
-        return reply.media;
-    }
-    if reply.timestamp_ms < entry.summary.latest_event_timestamp_ms.unwrap_or(0) {
-        return Vec::new();
+/// the reply the thread already ends on only fills in its preview and time. anything else
+/// counts once, when sync brought it and it's newer than a latest we know the time of. a
+/// summary from the event cache names its latest reply without the event, so its time is
+/// often unknown until that reply turns up
+fn add_reply(
+    entry: &mut ThreadListEntry,
+    reply: Normalised,
+    fresh: bool,
+    counted: &mut HashSet<OwnedEventId>,
+) -> Vec<MediaRegistration> {
+    let Normalised {
+        event_id,
+        timestamp_ms,
+        item,
+        media,
+    } = reply;
+    if entry.summary.latest_event_id.as_ref() != Some(&event_id) {
+        let newer = entry
+            .summary
+            .latest_event_timestamp_ms
+            .is_none_or(|latest| timestamp_ms > latest);
+        if !fresh || !newer || !counted.insert(event_id.clone()) {
+            return Vec::new();
+        }
+        entry.summary.reply_count = entry.summary.reply_count.saturating_add(1);
+        entry.summary.latest_event_id = Some(event_id);
+    } else {
+        counted.insert(event_id);
     }
-    entry.summary.reply_count = entry.summary.reply_count.saturating_add(1);
-    entry.summary.latest_event_id = Some(reply.event_id);
-    entry.summary.latest_event_timestamp_ms = Some(reply.timestamp_ms);
-    entry.latest_activity_ms = entry.latest_activity_ms.max(reply.timestamp_ms);
-    entry.latest_reply = Some(reply.item);
+    entry.summary.latest_event_timestamp_ms = Some(timestamp_ms);
+    entry.latest_activity_ms = entry.latest_activity_ms.max(timestamp_ms);
+    entry.latest_reply = Some(item);
     entry.root = with_summary(entry.root.clone(), &entry.summary);
-    reply.media
+    media
 }
 
 fn with_summary(mut root: TimelineItem, summary: &ThreadSummary) -> TimelineItem {
@@ -504,15 +702,27 @@ async fn fetch_root(
     room: &Room,
     root_event_id: &OwnedEventId,
     current_user_id: Option<OwnedUserId>,
-) -> Option<Built> {
-    let root = room
-        .load_or_fetch_event(root_event_id, None)
-        .await
-        .inspect_err(
-            |error| tracing::debug!(%root_event_id, %error, "fetching a thread root failed"),
-        )
-        .ok()?;
-    build_entry(room, root, current_user_id.as_deref()).await
+) -> Fetched {
+    let root = match room.load_or_fetch_event(root_event_id, None).await {
+        Ok(root) => root,
+        Err(source) => {
+            let error = Error::from_sdk("fetching a thread root failed", source);
+            tracing::debug!(%root_event_id, %error, "fetching a thread root failed");
+            let missing = error
+                .server
+                .as_ref()
+                .is_some_and(|server| server.status == 404);
+            return if missing {
+                Fetched::Missing
+            } else {
+                Fetched::Failed
+            };
+        }
+    };
+    match build_entry(room, root, current_user_id.as_deref()).await {
+        Some(built) => Fetched::Built(built),
+        None => Fetched::Missing,
+    }
 }
 
 async fn build_entry(
@@ -554,52 +764,63 @@ async fn build_entry(
     })
 }
 
+/// the sdk's corrected timestamp, which a sender's clock running ahead can't push past now
 async fn normalise(
     room: &Room,
     event: &SdkTimelineEvent,
     current_user_id: Option<&ruma::UserId>,
 ) -> Option<Normalised> {
     let event_id = event.event_id()?;
+    let timestamp_ms = u64::from(event.timestamp()?.get());
     let value: serde_json::Value = serde_json::from_str(event.raw().json().get()).ok()?;
     let identity = format!("{}\n{event_id}", room.room_id());
     let (mut item, media) = normalise_observed(&value, current_user_id, &identity)?;
     let TimelineItem::Event(normalised) = &mut item else {
         return None;
     };
+    normalised.timestamp_ms = timestamp_ms;
     normalised.highlight = event
         .push_actions()
         .is_some_and(|actions| actions.iter().any(Action::is_highlight));
-    if let Some(sender) = sender(room, &normalised.sender.user_id).await {
+    if let Some(sender) = member_sender(room, &normalised.sender.user_id).await {
         normalised.sender = sender;
     }
     // a preview doesn't chase what a root or a reply was replying to
     resolve_replies(std::iter::once(&mut **normalised), &HashMap::new());
     Some(Normalised {
         event_id,
-        timestamp_ms: normalised.timestamp_ms,
+        timestamp_ms,
         item,
         media,
     })
 }
 
-async fn sender(room: &Room, user_id: &OwnedUserId) -> Option<Sender> {
-    let member = room.get_member(user_id).await.ok().flatten()?;
-    Some(Sender {
-        user_id: user_id.clone(),
-        display_name: member.display_name().map(str::to_owned),
-        avatar: member.avatar_url().map(MediaId::plain),
-        name_ambiguous: member.name_ambiguous(),
-    })
+fn closure(state: RoomState) -> Option<ThreadListClosure> {
+    match state {
+        RoomState::Left => Some(ThreadListClosure::Left),
+        RoomState::Banned => Some(ThreadListClosure::Banned),
+        _ => None,
+    }
 }
 
-/// the room's event cache, followed for as long as the list is open. a lag loses whatever
-/// it skipped until the thread moves again
+/// the room's event cache and membership, followed for as long as the list is open
 async fn listen(
     room: Room,
     generation: u64,
     current_user_id: Option<OwnedUserId>,
     arrivals: mpsc::UnboundedSender<Arrival>,
 ) {
+    let room_id = room.room_id().to_owned();
+    let left = |reason| Arrival::Left {
+        room_id: room_id.clone(),
+        generation,
+        reason,
+    };
+    let mut info = room.subscribe_info();
+    if let Some(reason) = closure(room.state()) {
+        let _ = arrivals.send(left(reason));
+        return;
+    }
     let subscribed = async {
         let (cache, drop_handles) = room.event_cache().await?;
         let (_, subscriber) = cache.subscribe().await?;
@@ -609,16 +830,32 @@ async fn listen(
     let (_drop_handles, mut subscriber): (_, RoomEventCacheSubscriber) = match subscribed {
         Ok(subscribed) => subscribed,
         Err(error) => {
-            tracing::warn!(room_id = %room.room_id(), %error, "thread list can't follow the room");
+            tracing::warn!(%room_id, %error, "thread list can't follow the room");
             return;
         }
     };
-    let room_id = room.room_id().to_owned();
     loop {
-        let update = match subscriber.recv().await {
+        let update = tokio::select! {
+            update = subscriber.recv() => update,
+            Some(changed) = info.next() => {
+                if let Some(reason) = closure(changed.state()) {
+                    let _ = arrivals.send(left(reason));
+                    return;
+                }
+                continue;
+            }
+        };
+        let update = match update {
             Ok(update) => update,
             Err(RecvError::Lagged(skipped)) => {
-                tracing::warn!(%room_id, skipped, "thread list listener lagged");
+                tracing::debug!(%room_id, skipped, "thread list listener lagged, refreshing");
+                let lagged = Arrival::Lagged {
+                    room_id: room_id.clone(),
+                    generation,
+                };
+                if arrivals.send(lagged).is_err() {
+                    return;
+                }
                 continue;
             }
             Err(RecvError::Closed) => return,
@@ -626,14 +863,13 @@ async fn listen(
         let RoomEventCacheUpdate::UpdateTimelineEvents(update) = update else {
             continue;
         };
-        let paginated = matches!(update.origin, EventsOrigin::Pagination);
-        for event in added(update.diffs) {
+        let synced = matches!(update.origin, EventsOrigin::Sync);
+        for (event, inserted) in added(update.diffs) {
             let thread_root = extract_thread_root(event.raw());
             if thread_root.is_none() && event.thread_summary.summary().is_none() {
                 continue;
             }
-            let Some(normalised) =
-                normalise(&room, &event, current_user_id.as_deref()).await
+            let Some(normalised) = normalise(&room, &event, current_user_id.as_deref()).await
             else {
                 continue;
             };
@@ -641,7 +877,7 @@ async fn listen(
                 Some(root_event_id) => Live::Reply {
                     root_event_id,
                     reply: normalised,
-                    paginated,
+                    fresh: synced && inserted,
                 },
                 None => Live::Root(normalised),
             };
@@ -657,15 +893,21 @@ async fn listen(
     }
 }
 
-fn added(diffs: Vec<VectorDiff<SdkTimelineEvent>>) -> Vec<SdkTimelineEvent> {
+/// each event with whether it's new to the cache, rather than a replacement or a reset
+fn added(diffs: Vec<VectorDiff<SdkTimelineEvent>>) -> Vec<(SdkTimelineEvent, bool)> {
     let mut events = Vec::new();
     for diff in diffs {
         match diff {
-            VectorDiff::Append { values } | VectorDiff::Reset { values } => events.extend(values),
+            VectorDiff::Append { values } => {
+                events.extend(values.into_iter().map(|event| (event, true)));
+            }
+            VectorDiff::Reset { values } => {
+                events.extend(values.into_iter().map(|event| (event, false)));
+            }
             VectorDiff::PushBack { value }
             | VectorDiff::PushFront { value }
-            | VectorDiff::Insert { value, .. }
-            | VectorDiff::Set { value, .. } => events.push(value),
+            | VectorDiff::Insert { value, .. } => events.push((value, true)),
+            VectorDiff::Set { value, .. } => events.push((value, false)),
             VectorDiff::Clear
             | VectorDiff::PopBack
             | VectorDiff::PopFront
@@ -691,31 +933,33 @@ mod tests {
         }
     }
 
-    fn entry(root: &str, latest_activity_ms: u64) -> ThreadListEntry {
+    fn entry(root: &str, latest_timestamp_ms: Option<u64>) -> ThreadListEntry {
         ThreadListEntry {
             root_event_id: root.try_into().unwrap(),
             root: TimelineItem::DayDivider { timestamp_ms: 0 },
             summary: ThreadSummary {
                 reply_count: 2,
                 latest_event_id: Some(owned_event_id!("$latest")),
-                latest_event_timestamp_ms: Some(latest_activity_ms),
+                latest_event_timestamp_ms: latest_timestamp_ms,
             },
             latest_reply: None,
-            latest_activity_ms,
+            latest_activity_ms: latest_timestamp_ms.unwrap_or(1),
         }
     }
 
     #[test]
     fn a_reply_counts_once_and_only_when_newer() {
-        let mut thread = entry("$root", 20);
-        add_reply(&mut thread, reply("$latest", 20));
-        add_reply(&mut thread, reply("$older", 10));
+        let mut counted = HashSet::new();
+        let mut thread = entry("$root", Some(20));
+        add_reply(&mut thread, reply("$latest", 20), true, &mut counted);
+        add_reply(&mut thread, reply("$older", 10), true, &mut counted);
+        add_reply(&mut thread, reply("$replayed", 40), false, &mut counted);
         assert_eq!(
             thread.summary.reply_count, 2,
-            "a replayed or paginated reply was counted"
+            "a replayed or older reply was counted"
         );
-        add_reply(&mut thread, reply("$newer", 30));
-        add_reply(&mut thread, reply("$newer", 30));
+        add_reply(&mut thread, reply("$newer", 30), true, &mut counted);
+        add_reply(&mut thread, reply("$newer", 30), true, &mut counted);
         assert_eq!(thread.summary.reply_count, 3);
         assert_eq!(thread.latest_activity_ms, 30);
         assert_eq!(
@@ -725,12 +969,24 @@ mod tests {
     }
 
     #[test]
+    fn the_latest_reply_dates_a_summary_without_one() {
+        let mut counted = HashSet::new();
+        let mut thread = entry("$root", None);
+        add_reply(&mut thread, reply("$latest", 50), false, &mut counted);
+        assert_eq!(thread.summary.reply_count, 2);
+        assert_eq!(thread.latest_activity_ms, 50);
+        add_reply(&mut thread, reply("$latest", 50), true, &mut counted);
+        add_reply(&mut thread, reply("$replayed", 60), false, &mut counted);
+        assert_eq!(thread.summary.reply_count, 2, "a replay moved the count");
+    }
+
+    #[test]
     fn upserts_keep_newest_activity_first() {
-        let mut entries = imbl::Vector::new();
+        let mut entries = Vector::new();
         for (root, activity) in [("$a", 10), ("$b", 30), ("$c", 20), ("$d", 20)] {
-            ThreadListEntry::upsert(&mut entries, entry(root, activity));
+            ThreadListEntry::upsert(&mut entries, entry(root, Some(activity)));
         }
-        ThreadListEntry::upsert(&mut entries, entry("$a", 40));
+        ThreadListEntry::upsert(&mut entries, entry("$a", Some(40)));
         let order: Vec<_> = entries
             .iter()
             .map(|entry| entry.root_event_id.as_str())

Modified file: src/engine/watchers.rs+99 additions−42 deletions@@ -1,5 +1,5 @@
 use std::{
-    collections::{BTreeMap, HashMap, HashSet, VecDeque},
+    collections::{HashMap, HashSet, VecDeque},
     sync::Arc,
     time::Duration,
 };
@@ -7,6 +7,8 @@ use std::{
 use futures_util::StreamExt;
 use matrix_sdk::{
     Client, Room,
+    event_handler::EventHandlerHandle,
+    serde_helpers::extract_thread_root,
     ruma::events::{AnyMessageLikeEventContent, room::message::MessageType},
     send_queue::{LocalEchoContent, RoomSendQueueUpdate, SendQueueUpdate},
 };
@@ -16,19 +18,23 @@ use ruma::{
     OwnedEventId, OwnedRoomId, OwnedTransactionId,
     events::{
         AnyGlobalAccountDataEvent, AnyRoomAccountDataEvent, AnySyncStateEvent,
-        receipt::SyncReceiptEvent,
+        AnySyncTimelineEvent, receipt::SyncReceiptEvent,
     },
     serde::Raw,
     time::Instant,
 };
-use tokio::sync::{Notify, broadcast::error::RecvError, mpsc, watch};
+use tokio::sync::{
+    Notify,
+    broadcast::error::{RecvError, TryRecvError},
+    mpsc, watch,
+};
 
 use super::executor::TaskSet;
 use crate::{
-    ConnectionState, Event, ThreadUnread, UploadOutcome,
+    ConnectionState, Event, UploadOutcome,
     encryption::encryption_state,
     error::classify_sdk_error,
-    normalise::{ENABLED_ROOMS, ROOM_PACKS, USER_PACK, thread_unreads},
+    normalise::{ENABLED_ROOMS, ROOM_PACKS, USER_PACK, recount_thread_unreads},
     space_order,
     transport::EventSender,
 };
@@ -436,71 +442,122 @@ pub(super) async fn rooms(client: Client, events: EventSender) {
             }
         }),
     ];
-    let (receipts, mut receipt_rooms) = mpsc::unbounded_channel();
-    let receipt_handler = client.add_event_handler(move |_: SyncReceiptEvent, room: Room| {
-        let _ = receipts.send(room.room_id().to_owned());
-        async {}
-    });
+    let (touches, mut touched) = mpsc::unbounded_channel();
+    let thread_handlers = thread_handlers(&client, &touches);
     let mut timelines = client.event_cache().subscribe_to_room_generic_updates();
-    let mut thread_unreads = ThreadUnreads::default();
     'invalidate: loop {
         if events.send(Event::RoomsChanged).await.is_err() {
             break;
         }
         // thread events and threaded receipts leave room info alone, so neither reaches `info`
+        let mut pending = Touched::default();
         loop {
-            let room_id = tokio::select! {
-                update = rooms.next() => if update.is_none() { break 'invalidate } else { break },
+            let mut invalidate = tokio::select! {
+                update = rooms.next() => if update.is_none() { break 'invalidate } else { true },
                 update = info.recv() => {
                     if matches!(update, Err(RecvError::Closed)) { break 'invalidate }
-                    break;
+                    true
+                }
+                () = changed.notified() => true,
+                Some((room_id, root)) = touched.recv() => {
+                    pending.add(room_id, root);
+                    false
                 }
-                () = changed.notified() => break,
-                Some(room_id) = receipt_rooms.recv() => room_id,
                 update = timelines.recv() => match update {
-                    Ok(update) => update.room_id,
-                    Err(RecvError::Lagged(_)) => {
-                        thread_unreads.0.clear();
-                        break;
+                    Ok(update) => {
+                        pending.add(update.room_id, None);
+                        false
                     }
+                    Err(RecvError::Lagged(_)) => true,
                     Err(RecvError::Closed) => break 'invalidate,
                 },
             };
-            if thread_unreads.changed(&client, room_id).await {
+            // a sync's room info lands before the event cache has stored its events, so
+            // give it a moment and take the whole sync as one invalidation
+            matrix_sdk::sleep::sleep(SYNC_SETTLE).await;
+            loop {
+                match info.try_recv() {
+                    Ok(_) | Err(TryRecvError::Lagged(_)) => invalidate = true,
+                    Err(_) => break,
+                }
+            }
+            while let Ok((room_id, root)) = touched.try_recv() {
+                pending.add(room_id, root);
+            }
+            loop {
+                match timelines.try_recv() {
+                    Ok(update) => pending.add(update.room_id, None),
+                    Err(TryRecvError::Lagged(_)) => invalidate = true,
+                    Err(_) => break,
+                }
+            }
+            for (room_id, roots) in pending.0.drain() {
+                if let Some(room) = client.get_room(&room_id)
+                    && recount_thread_unreads(&room, roots).await.is_some()
+                {
+                    invalidate = true;
+                }
+            }
+            if invalidate {
                 break;
             }
         }
-        // a sync touching fifty rooms is one invalidation, not fifty
-        while let Ok(_) | Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) =
-            info.try_recv()
-        {}
     }
-    for handler in handlers.into_iter().chain([receipt_handler]) {
+    for handler in handlers.into_iter().chain(thread_handlers) {
         client.remove_event_handler(handler);
     }
 }
 
+const SYNC_SETTLE: Duration = Duration::from_millis(100);
+
+/// rooms to recount, with the threads a sync brought news of
 #[derive(Default)]
-struct ThreadUnreads(HashMap<OwnedRoomId, BTreeMap<OwnedEventId, ThreadUnread>>);
+struct Touched(HashMap<OwnedRoomId, HashSet<OwnedEventId>>);
 
-impl ThreadUnreads {
-    async fn changed(&mut self, client: &Client, room_id: OwnedRoomId) -> bool {
-        let Some(room) = client.get_room(&room_id) else {
-            return self.0.remove(&room_id).is_some();
-        };
-        let unreads = thread_unreads(&room).await;
-        if self
-            .0
-            .get(&room_id)
-            .map_or(unreads.is_empty(), |known| *known == unreads)
-        {
-            return false;
-        }
-        self.0.insert(room_id, unreads);
-        true
+impl Touched {
+    fn add(&mut self, room_id: OwnedRoomId, root: Option<OwnedEventId>) {
+        let roots = self.0.entry(room_id).or_default();
+        roots.extend(root);
     }
 }
 
+/// a thread event names its thread before the event cache has it, and only the user's own
+/// receipts move their counts
+fn thread_handlers(
+    client: &Client,
+    touches: &mpsc::UnboundedSender<(OwnedRoomId, Option<OwnedEventId>)>,
+) -> [EventHandlerHandle; 2] {
+    let own_user_id = client.user_id().map(ToOwned::to_owned);
+    [
+        client.add_event_handler({
+            let touches = touches.clone();
+            move |event: Raw<AnySyncTimelineEvent>, room: Room| {
+                if let Some(root) = extract_thread_root(&event) {
+                    let _ = touches.send((room.room_id().to_owned(), Some(root)));
+                }
+                async {}
+            }
+        }),
+        client.add_event_handler({
+            let touches = touches.clone();
+            move |event: SyncReceiptEvent, room: Room| {
+                let own = own_user_id.as_ref().is_some_and(|own_user_id| {
+                    event
+                        .content
+                        .0
+                        .values()
+                        .flat_map(|receipts| receipts.values())
+                        .any(|users| users.contains_key(own_user_id))
+                });
+                if own {
+                    let _ = touches.send((room.room_id().to_owned(), None));
+                }
+                async {}
+            }
+        }),
+    ]
+}
+
 /// packs live in room state and account data, and leaving a room takes its packs with it
 pub(super) async fn image_packs(client: Client, events: EventSender) {
     let changed = Arc::new(Notify::new());

Modified file: src/lib.rs+2 additions−1 deletions@@ -68,7 +68,8 @@ pub use model::{
     RecentHistoryState, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText,
     RoomCategory, RoomCommand, RoomHero, RoomListFilter, RoomMatcher, RoomMembership,
     RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomPreset, RoomSummary,
-    RoomVisibility, SendState, Sender, SpaceChild, Termination, ThreadListCommand, ThreadListEntry,
+    RoomVisibility, SendState, Sender, SpaceChild, Termination, ThreadListClosure,
+    ThreadListCommand, ThreadListEntry,
     ThreadSummary, ThreadUnread, TimelineCommand, TimelineContent, TimelineDiff, TimelineDiffError,
     TimelineEvent, TimelineId, TimelineItem, TimelineItemId, TypingState, UploadDetails,
     UploadOutcome, UploadThumbnail, UtdCause, VerificationAction, VerificationFailure,

Modified file: src/mirror.rs+5 additions−4 deletions@@ -21,7 +21,8 @@ use tokio::sync::broadcast;
 use crate::{
     ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
     EventContextCloseReason, HistoryRepairReport, HistoryState, ImagePacks, RecentHistoryState,
-    RoomSummary, ThreadListEntry, TimelineCommand, TimelineDiff, TimelineId, TimelineItem, TypingState,
+    RoomSummary, ThreadListClosure, ThreadListEntry, TimelineCommand, TimelineDiff, TimelineId,
+    TimelineItem, TypingState,
     UploadOutcome, VerificationState, VerificationTarget, engine::executor::EngineBoxFuture,
     transport::PublicationRegistration,
 };
@@ -200,7 +201,7 @@ pub enum ThreadListChange {
         entries: Vector<ThreadListEntry>,
     },
     History(HistoryState),
-    Closed,
+    Closed(ThreadListClosure),
 }
 
 /// drains the event stream into a `State` that any number of `MirrorHandle`s watch
@@ -482,11 +483,11 @@ impl Mirror {
                     change: ThreadListChange::History(history),
                 })
             }),
-            Event::ThreadListClosed { room_id } => self.publish(|state| {
+            Event::ThreadListClosed { room_id, reason } => self.publish(|state| {
                 state.thread_lists.remove(&room_id)?;
                 Some(StateChange::ThreadList {
                     room_id,
-                    change: ThreadListChange::Closed,
+                    change: ThreadListChange::Closed(reason),
                 })
             }),
             Event::TypingChanged(typing) => self.publish(|state| {

Modified file: src/model.rs+11 additions@@ -982,6 +982,7 @@ pub enum Event {
     /// nothing more comes for it until another `ThreadListReset`
     ThreadListClosed {
         room_id: OwnedRoomId,
+        reason: ThreadListClosure,
     },
     /// a room that stops being watched says so with an empty `users`
     TypingChanged(TypingState),
@@ -1914,6 +1915,16 @@ pub struct ThreadSummary {
     pub latest_event_timestamp_ms: Option<u64>,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum ThreadListClosure {
+    /// `Close` asked for it
+    Requested,
+    /// left or kicked
+    Left,
+    Banned,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[non_exhaustive]

Modified file: src/normalise/member.rs+13 additions−3 deletions@@ -1,7 +1,7 @@
-use matrix_sdk::room::RoomMember;
-use ruma::events::room::member::MembershipState;
+use matrix_sdk::{Room, room::RoomMember};
+use ruma::{OwnedUserId, events::room::member::MembershipState};
 
-use crate::{MediaId, MemberMembership, MemberSummary};
+use crate::{MediaId, MemberMembership, MemberSummary, Sender};
 
 pub(crate) fn member_summary(member: RoomMember) -> MemberSummary {
     MemberSummary {
@@ -23,3 +23,13 @@ pub(crate) fn member_summary(member: RoomMember) -> MemberSummary {
         service: member.is_service_member(),
     }
 }
+
+pub(crate) async fn member_sender(room: &Room, user_id: &OwnedUserId) -> Option<Sender> {
+    let member = room.get_member(user_id).await.ok().flatten()?;
+    Some(Sender {
+        user_id: user_id.clone(),
+        display_name: member.display_name().map(str::to_owned),
+        avatar: member.avatar_url().map(MediaId::plain),
+        name_ambiguous: member.name_ambiguous(),
+    })
+}

Modified file: src/normalise/mod.rs+2 additions−2 deletions@@ -15,9 +15,9 @@ pub(crate) use image_pack::{
 };
 pub(crate) use link_preview::{ROOM_PREVIEW_URLS, server_preview, url_previews_allowed};
 pub(crate) use media::{MediaRegistration, MediaScope, MediaSourceInfo};
-pub(crate) use member::member_summary;
+pub(crate) use member::{member_sender, member_summary};
 pub(crate) use observed::normalise_observed;
 pub(crate) use room::notification_mode;
 pub(crate) use room::{room_summaries, space_child, stored_space_order};
-pub(crate) use thread::thread_unreads;
+pub(crate) use thread::{recount_thread_unreads, stored_thread_unreads};
 pub(crate) use timeline::{message_kind, normalise_timeline, normalise_timeline_diffs};

Modified file: src/normalise/room.rs+1−1

@@ -62,7 +62,7 @@ async fn room_summary(room: &Room, spaces: &SpaceService) -> RoomSummary {

6262 } else {

6363 Vec::new()

6464 };

65- let thread_unreads = super::thread_unreads(room).await;

65+ let thread_unreads = super::stored_thread_unreads(room).await;

6666 RoomSummary {

6767 room_id: room.room_id().to_owned(),

6868 name,

Modified file: src/normalise/thread.rs+161 additions−75 deletions@@ -1,115 +1,157 @@
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashSet};
 
 use matrix_sdk::{
-    Room, RoomState, deserialized_responses::TimelineEvent, serde_helpers::extract_thread_root,
+    Room, RoomState, cross_process_lock::MappedCrossProcessLockState,
+    deserialized_responses::TimelineEvent,
 };
+use matrix_sdk_base::event_cache::store::EventCacheStoreLockGuard;
 use ruma::{
-    OwnedEventId, UserId,
-    events::receipt::{ReceiptThread, ReceiptType},
+    OwnedEventId,
+    events::{
+        receipt::{ReceiptThread, ReceiptType},
+        relation::RelationType,
+    },
 };
-use serde_json::Value;
+use serde_json::{Value, json};
 
 use crate::ThreadUnread;
 
-struct Reply {
-    event_id: Option<OwnedEventId>,
-    own: bool,
-    counts: bool,
-    notify: bool,
-    highlight: bool,
+fn key(room: &Room) -> Vec<u8> {
+    format!("macltok.thread_unreads.{}", room.room_id()).into_bytes()
 }
 
-impl Reply {
-    fn of(event: &TimelineEvent, own_user_id: &UserId) -> Self {
-        let own = event.sender().as_deref() == Some(own_user_id);
-        let counts = !own && counts_as_unread(event);
-        let actions = event.push_actions().unwrap_or_default();
-        Self {
-            event_id: event.event_id(),
-            own,
-            counts,
-            notify: counts && actions.iter().any(|action| action.should_notify()),
-            highlight: counts && actions.iter().any(|action| action.is_highlight()),
+/// what the last recount left, so a room list never touches the event cache
+pub(crate) async fn stored_thread_unreads(room: &Room) -> BTreeMap<OwnedEventId, ThreadUnread> {
+    if room.state() != RoomState::Joined {
+        return BTreeMap::new();
+    }
+    match room.client().state_store().get_custom_value(&key(room)).await {
+        Ok(Some(bytes)) => decode(&bytes),
+        Ok(None) => BTreeMap::new(),
+        Err(error) => {
+            tracing::debug!(room_id = %room.room_id(), %error, "reading thread unreads failed");
+            BTreeMap::new()
         }
     }
 }
 
 /// the sdk leaves thread events out of the room's counts once threads are on and keeps none
-/// of its own for them. so these are worked out the same way over the events it has in
-/// memory, newest first until a receipt for that thread (msc3771), an unthreaded receipt or
-/// the user's own reply. a receipt older than memory counts everything in memory, as the
-/// sdk's does
-pub(crate) async fn thread_unreads(room: &Room) -> BTreeMap<OwnedEventId, ThreadUnread> {
-    if room.state() != RoomState::Joined {
-        return BTreeMap::new();
+/// of its own for them. this counts `touched` and every thread that still had something
+/// unread from the event cache store, which holds every reply it ever took in whatever is
+/// in memory, against a receipt for that thread (msc3771), an unthreaded one or the user's
+/// own reply. `None` when nothing moved
+pub(crate) async fn recount_thread_unreads(
+    room: &Room,
+    touched: HashSet<OwnedEventId>,
+) -> Option<BTreeMap<OwnedEventId, ThreadUnread>> {
+    let known = stored_thread_unreads(room).await;
+    let mut roots = touched;
+    roots.extend(known.keys().cloned());
+    if roots.is_empty() || room.state() != RoomState::Joined {
+        return None;
     }
-    let own_user_id = room.own_user_id();
-    let unthreaded = own_receipts(room, ReceiptThread::Unthreaded).await;
-    let Ok((cache, _drop_handles)) = room.event_cache().await else {
-        return BTreeMap::new();
+    let store = match room.client().event_cache_store().lock().await {
+        Ok(MappedCrossProcessLockState::Clean(store) | MappedCrossProcessLockState::Dirty(store)) => {
+            store
+        }
+        Err(error) => {
+            tracing::debug!(room_id = %room.room_id(), %error, "locking the event cache store failed");
+            return None;
+        }
     };
-    let mut threads: HashMap<OwnedEventId, Vec<Reply>> = HashMap::new();
-    let scanned = cache
-        .rfind_map_event_in_memory_by(|event| {
-            let event_id = event.event_id();
-            if event_id.is_some_and(|event_id| unthreaded.contains(&event_id)) {
-                return Some(());
+    let own_user_id = room.own_user_id();
+    let unthreaded = read_through(room, &store, ReceiptThread::Unthreaded).await;
+    let mut counted = BTreeMap::new();
+    for root in roots {
+        let replies = match store
+            .find_event_relations(room.room_id(), &root, Some(&[RelationType::Thread][..]))
+            .await
+        {
+            Ok(replies) => replies,
+            Err(error) => {
+                tracing::debug!(room_id = %room.room_id(), %root, %error, "reading a thread failed");
+                if let Some(unread) = known.get(&root) {
+                    counted.insert(root, *unread);
+                }
+                continue;
             }
-            if let Some(root) = extract_thread_root(event.raw()) {
-                threads
-                    .entry(root)
-                    .or_default()
-                    .push(Reply::of(event, own_user_id));
+        };
+        let mut read = unthreaded.max(read_through(room, &store, ReceiptThread::Thread(root.clone())).await);
+        for (reply, _) in &replies {
+            if reply.sender().as_deref() == Some(own_user_id) {
+                read = read.max(timestamp_ms(reply));
             }
-            None
-        })
-        .await;
-    if let Err(error) = scanned {
-        tracing::debug!(room_id = %room.room_id(), %error, "reading events for thread unreads failed");
-        return BTreeMap::new();
-    }
-    let mut unreads = BTreeMap::new();
-    for (root, replies) in threads {
-        let read = own_receipts(room, ReceiptThread::Thread(root.clone())).await;
+        }
         let mut unread = ThreadUnread::default();
-        for reply in replies {
-            if reply.own
-                || reply
-                    .event_id
-                    .as_ref()
-                    .is_some_and(|event_id| read.contains(event_id))
+        for (reply, _) in &replies {
+            let Some(at) = timestamp_ms(reply) else {
+                continue;
+            };
+            if read.is_some_and(|read| at <= read)
+                || reply.sender().as_deref() == Some(own_user_id)
+                || !counts_as_unread(reply)
             {
-                break;
+                continue;
             }
-            unread.messages += u64::from(reply.counts);
-            unread.notifications += u64::from(reply.notify);
-            unread.mentions += u64::from(reply.highlight);
+            let actions = reply.push_actions().unwrap_or_default();
+            unread.messages += 1;
+            unread.notifications += u64::from(actions.iter().any(|action| action.should_notify()));
+            unread.mentions += u64::from(actions.iter().any(|action| action.is_highlight()));
         }
         if unread != ThreadUnread::default() {
-            unreads.insert(root, unread);
+            counted.insert(root, unread);
         }
     }
-    unreads
+    if counted == known {
+        return None;
+    }
+    let stored = room
+        .client()
+        .state_store()
+        .set_custom_value(&key(room), encode(&counted))
+        .await;
+    if let Err(error) = stored {
+        tracing::debug!(room_id = %room.room_id(), %error, "saving thread unreads failed");
+    }
+    Some(counted)
 }
 
-async fn own_receipts(room: &Room, thread: ReceiptThread) -> Vec<OwnedEventId> {
-    let mut found = Vec::new();
+/// receipts sit on events, and stream order between a thread and the rest of the room is
+/// gone once it's in the store, so the event's timestamp stands for its place. a receipt
+/// on an event the store never had falls back to when it was sent
+async fn read_through(
+    room: &Room,
+    store: &EventCacheStoreLockGuard,
+    thread: ReceiptThread,
+) -> Option<u64> {
+    let mut latest = None;
     for receipt_type in [ReceiptType::Read, ReceiptType::ReadPrivate] {
-        match room
+        let (event_id, receipt) = match room
             .load_user_receipt(receipt_type, thread.clone(), room.own_user_id())
             .await
         {
-            Ok(Some((event_id, _))) => found.push(event_id),
-            Ok(None) => {}
+            Ok(Some(found)) => found,
+            Ok(None) => continue,
             Err(error) => {
                 tracing::debug!(room_id = %room.room_id(), %error, "reading a receipt failed");
+                continue;
             }
+        };
+        let at = match store.find_event(room.room_id(), &event_id).await {
+            Ok(Some(event)) => timestamp_ms(&event),
+            _ => None,
         }
+        .or_else(|| receipt.ts.map(|ts| u64::from(ts.get())));
+        latest = latest.max(at);
     }
-    found
+    latest
 }
 
-/// the sdk's rule for its room counts, which is private to it
+fn timestamp_ms(event: &TimelineEvent) -> Option<u64> {
+    event.timestamp().map(|ts| u64::from(ts.get()))
+}
+
+/// the sdk's `marks_as_unread` for its room counts, which is private to it
 fn counts_as_unread(event: &TimelineEvent) -> bool {
     let raw = event.raw();
     let redacted = raw
@@ -117,10 +159,22 @@ fn counts_as_unread(event: &TimelineEvent) -> bool {
         .ok()
         .flatten()
         .is_some_and(|unsigned| unsigned.get("redacted_because").is_some());
+    let edit = raw
+        .get_field::<Value>("content")
+        .ok()
+        .flatten()
+        .and_then(|content| {
+            content
+                .pointer("/m.relates_to/rel_type")
+                .and_then(Value::as_str)
+                .map(|rel_type| rel_type == "m.replace")
+        })
+        .unwrap_or(false);
     let message_like = matches!(
         raw.get_field::<String>("type").ok().flatten().as_deref(),
         Some(
-            "m.room.message"
+            "m.message"
+                | "m.room.message"
                 | "m.room.encrypted"
                 | "m.sticker"
                 | "m.poll.start"
@@ -129,5 +183,37 @@ fn counts_as_unread(event: &TimelineEvent) -> bool {
                 | "org.matrix.msc3381.poll.end"
         )
     );
-    message_like && !redacted
+    message_like && !edit && !redacted
+}
+
+fn encode(unreads: &BTreeMap<OwnedEventId, ThreadUnread>) -> Vec<u8> {
+    let value: serde_json::Map<String, Value> = unreads
+        .iter()
+        .map(|(root, unread)| {
+            (
+                root.to_string(),
+                json!([unread.messages, unread.mentions, unread.notifications]),
+            )
+        })
+        .collect();
+    serde_json::to_vec(&value).unwrap_or_default()
+}
+
+fn decode(bytes: &[u8]) -> BTreeMap<OwnedEventId, ThreadUnread> {
+    let Ok(value) = serde_json::from_slice::<BTreeMap<OwnedEventId, [u64; 3]>>(bytes) else {
+        return BTreeMap::new();
+    };
+    value
+        .into_iter()
+        .map(|(root, [messages, mentions, notifications])| {
+            (
+                root,
+                ThreadUnread {
+                    messages,
+                    mentions,
+                    notifications,
+                },
+            )
+        })
+        .collect()
 }

Modified file: tests/state.rs+19 additions−3 deletions@@ -7,6 +7,7 @@ use macltok::{
     Termination, ThreadUnread,
 };
 use matrix_sdk::{
+    ThreadingSupport,
     ruma::{
         RoomId,
         api::client::sync::sync_events::v5,
@@ -254,15 +255,30 @@ async fn thread_unreads_follow_threaded_receipts() {
         .ok(v5::Response::new("1".to_owned()))
         .mount()
         .await;
-    let client = server.client_builder().build().await;
+    let client = server
+        .client_builder()
+        .on_builder(|builder| {
+            builder.with_threading_support(ThreadingSupport::Enabled {
+                with_subscriptions: false,
+            })
+        })
+        .build()
+        .await;
     let room_id = room_id!("!threads:localhost");
+    let own = client.user_id().unwrap().to_owned();
     server.mock_room_state_encryption().plain().mount().await;
-    server.sync_joined_room(&client, room_id).await;
     client.event_cache().subscribe().unwrap();
+    // without our own member event there's no push context, so no push actions to count
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_state_event(EventFactory::new().room(room_id).member(&own)),
+        )
+        .await;
     let config = EngineConfig::default().with_initial_history_limit(0);
     let (engine, handle, mut events) = Engine::from_client(client.clone(), config).await.unwrap();
     let running = tokio::spawn(engine.run());
-    let own = client.user_id().unwrap().to_owned();
 
     let (root, first, second) = (
         event_id!("$root"),

macltok

Commit 21b9227729a4

Commit `21b9227729a4`