macltok

public

Star0

Fork 0

Watch0

99 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `e7c1b037154d`

tarball

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

3 files changed+249-57

threads: count a page's replies only once they postdate it

continuwuity 26.9 bundles a thread's latest reply in /threads as bare content, so a page entry had no dated latest, and every reply the event cache re-inserted under a sync read as newer and counted on top of the server's count, doubling it. add_reply now also requires a reply to postdate when its thread's count was read off the server, kept per root in Tally, and a newer reply that doesn't count still becomes the latest.

Modified file: docs/design-decisions.md+10 additions−6 deletions@@ -138,12 +138,16 @@ 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, 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
+counts once, by id, when sync just brought it, it's newer than the thread's
+latest and it was sent after the count was read off the server, all by the sdk's
+corrected timestamp. the latest alone can't date a count. a summary from the
+event cache names its latest reply without the event, and continuwuity 26.9
+bundles the latest as bare content, so with only that check every reply the
+cache re-inserted read as new and a page's count doubled. a newer reply that
+doesn't count still becomes the latest. 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

Modified file: src/engine/thread_list.rs+129 additions−49 deletions@@ -97,6 +97,16 @@ struct Page {
 struct Built {
     entry: ThreadListEntry,
     media: Vec<MediaRegistration>,
+    /// when its count was read off the server, for an entry fetched just now
+    read_at: Option<u64>,
+}
+
+#[derive(Default)]
+struct Tally {
+    /// replies already counted, so the same event synced twice counts once
+    counted: HashSet<OwnedEventId>,
+    /// when each thread's count was read off the server, which had every reply sent before
+    read_at: HashMap<OwnedEventId, u64>,
 }
 
 /// what a list is loading, and who waits on it
@@ -118,8 +128,7 @@ struct List {
     page_limit: u16,
     load: Load,
     refresh_owed: bool,
-    /// replies already counted, so the same event synced twice counts once
-    counted: HashSet<OwnedEventId>,
+    tally: Tally,
     /// roots being fetched or waiting for a slot, with the replies that arrived meanwhile
     fetching: HashMap<OwnedEventId, Vec<Normalised>>,
     waiting_roots: VecDeque<OwnedEventId>,
@@ -227,7 +236,7 @@ impl ThreadLists {
                 page_limit: DEFAULT_PAGE,
                 load: Load::Idle,
                 refresh_owed: false,
-                counted: HashSet::new(),
+                tally: Tally::default(),
                 fetching: HashMap::new(),
                 waiting_roots: VecDeque::new(),
                 fetches_in_flight: 0,
@@ -449,9 +458,14 @@ impl ThreadLists {
             } => {
                 if let Some(index) = list.position(&root_event_id) {
                     let mut entry = list.entries[index].clone();
-                    let media = add_reply(&mut entry, reply, fresh, &mut list.counted);
+                    let media = add_reply(&mut entry, reply, fresh, &mut list.tally);
                     if entry != list.entries[index] {
-                        self.publish(&room_id, vec![Built { entry, media }]).await;
+                        let built = Built {
+                            entry,
+                            media,
+                            read_at: None,
+                        };
+                        self.publish(&room_id, vec![built]).await;
                     }
                     return;
                 }
@@ -472,8 +486,12 @@ impl ThreadLists {
                 let mut entry = list.entries[index].clone();
                 entry.root = with_summary(root.item, &entry.summary);
                 if entry != list.entries[index] {
-                    let media = root.media;
-                    self.publish(&room_id, vec![Built { entry, media }]).await;
+                    let built = Built {
+                        entry,
+                        media: root.media,
+                        read_at: None,
+                    };
+                    self.publish(&room_id, vec![built]).await;
                 }
             }
         }
@@ -525,16 +543,18 @@ impl ThreadLists {
                 let mut built = Built {
                     entry: entry.clone(),
                     media: Vec::new(),
+                    read_at: None,
                 };
                 for reply in replies {
-                    let media = add_reply(&mut built.entry, reply, true, &mut list.counted);
+                    let media = add_reply(&mut built.entry, reply, true, &mut list.tally);
                     built.media.extend(media);
                 }
                 (built.entry != entry).then_some(built)
             }
             (Fetched::Built(mut built), None) => {
+                list.tally.note(&built);
                 for reply in replies {
-                    let media = add_reply(&mut built.entry, reply, true, &mut list.counted);
+                    let media = add_reply(&mut built.entry, reply, true, &mut list.tally);
                     built.media.extend(media);
                 }
                 Some(built)
@@ -562,14 +582,11 @@ impl ThreadLists {
         };
         let mut media = Vec::new();
         let mut upserts = Vec::with_capacity(built.len());
-        for Built {
-            entry,
-            media: registrations,
-        } in built
-        {
-            media.extend(registrations);
-            ThreadListEntry::upsert(&mut list.entries, entry.clone());
-            upserts.push(entry);
+        for built in built {
+            list.tally.note(&built);
+            media.extend(built.media);
+            ThreadListEntry::upsert(&mut list.entries, built.entry.clone());
+            upserts.push(built.entry);
         }
         let scope = list.scope.clone();
         self.media.register(&scope, media).await;
@@ -588,13 +605,10 @@ impl ThreadLists {
         };
         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);
+        for built in built {
+            list.tally.note(&built);
+            media.extend(built.media);
+            ThreadListEntry::upsert(&mut list.entries, built.entry);
         }
         let entries = list.entries.clone();
         let scope = list.scope.clone();
@@ -623,6 +637,18 @@ impl List {
     }
 }
 
+impl Tally {
+    fn note(&mut self, built: &Built) {
+        if let Some(read_at) = built.read_at {
+            let known = self
+                .read_at
+                .entry(built.entry.root_event_id.clone())
+                .or_default();
+            *known = (*known).max(read_at);
+        }
+    }
+}
+
 fn current<'a>(
     lists: &'a mut HashMap<OwnedRoomId, List>,
     room_id: &OwnedRoomId,
@@ -637,15 +663,16 @@ fn not_open() -> Error {
     Error::new(ErrorKind::TimelineNotOpen, "thread list is not open")
 }
 
-/// 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
+/// a reply newer than the thread's latest becomes its latest, and counts once if sync just
+/// brought it and it was sent after the count was read off the server. the latest's time
+/// alone can't tell, since a summary can arrive without a usable latest event, an event
+/// cache summary names one without the event and continuwuity 26.9 bundles only its content,
+/// and then every reply the cache re-inserts reads as newer
 fn add_reply(
     entry: &mut ThreadListEntry,
     reply: Normalised,
     fresh: bool,
-    counted: &mut HashSet<OwnedEventId>,
+    tally: &mut Tally,
 ) -> Vec<MediaRegistration> {
     let Normalised {
         event_id,
@@ -653,18 +680,25 @@ fn add_reply(
         item,
         media,
     } = reply;
-    if entry.summary.latest_event_id.as_ref() != Some(&event_id) {
-        let newer = entry
+    if entry.summary.latest_event_id.as_ref() == Some(&event_id) {
+        tally.counted.insert(event_id);
+    } else {
+        if entry
             .summary
             .latest_event_timestamp_ms
-            .is_none_or(|latest| timestamp_ms > latest);
-        if !fresh || !newer || !counted.insert(event_id.clone()) {
+            .is_some_and(|latest| timestamp_ms <= latest)
+        {
             return Vec::new();
         }
-        entry.summary.reply_count = entry.summary.reply_count.saturating_add(1);
+        let read_at = tally
+            .read_at
+            .get(&entry.root_event_id)
+            .copied()
+            .unwrap_or(0);
+        if fresh && timestamp_ms > read_at && tally.counted.insert(event_id.clone()) {
+            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.latest_event_timestamp_ms = Some(timestamp_ms);
     entry.latest_activity_ms = entry.latest_activity_ms.max(timestamp_ms);
@@ -687,6 +721,7 @@ async fn fetch_page(
     current_user_id: Option<OwnedUserId>,
     media_prefix: String,
 ) -> Result<Page, Error> {
+    let read_at = now_ms();
     let roots = room
         .list_threads(ListThreadsOptions {
             from,
@@ -697,8 +732,14 @@ async fn fetch_page(
         .map_err(|source| Error::from_sdk("listing threads failed", source))?;
     let mut entries = Vec::with_capacity(roots.chunk.len());
     for root in roots.chunk {
-        if let Some(built) =
-            build_entry(&room, root, current_user_id.as_deref(), &media_prefix).await
+        if let Some(built) = build_entry(
+            &room,
+            root,
+            current_user_id.as_deref(),
+            &media_prefix,
+            read_at,
+        )
+        .await
         {
             entries.push(built);
         }
@@ -715,6 +756,7 @@ async fn fetch_root(
     current_user_id: Option<OwnedUserId>,
     media_prefix: &str,
 ) -> Fetched {
+    let read_at = now_ms();
     let root = match room.load_or_fetch_event(root_event_id, None).await {
         Ok(root) => root,
         Err(source) => {
@@ -731,7 +773,15 @@ async fn fetch_root(
             };
         }
     };
-    match build_entry(room, root, current_user_id.as_deref(), media_prefix).await {
+    match build_entry(
+        room,
+        root,
+        current_user_id.as_deref(),
+        media_prefix,
+        read_at,
+    )
+    .await
+    {
         Some(built) => Fetched::Built(built),
         None => Fetched::Missing,
     }
@@ -742,6 +792,7 @@ async fn build_entry(
     root: SdkTimelineEvent,
     current_user_id: Option<&ruma::UserId>,
     media_prefix: &str,
+    read_at: u64,
 ) -> Option<Built> {
     let summary = root.thread_summary.summary().cloned();
     let latest = root.bundled_latest_thread_event.clone();
@@ -774,9 +825,14 @@ async fn build_entry(
             latest_reply,
         },
         media,
+        read_at: Some(read_at),
     })
 }
 
+fn now_ms() -> u64 {
+    u64::from(ruma::MilliSecondsSinceUnixEpoch::now().get())
+}
+
 /// the sdk's corrected timestamp, which a sender's clock running ahead can't push past now
 async fn normalise(
     room: &Room,
@@ -964,17 +1020,17 @@ mod tests {
 
     #[test]
     fn a_reply_counts_once_and_only_when_newer() {
-        let mut counted = HashSet::new();
+        let mut tally = Tally::default();
         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);
+        add_reply(&mut thread, reply("$latest", 20), true, &mut tally);
+        add_reply(&mut thread, reply("$older", 10), true, &mut tally);
+        add_reply(&mut thread, reply("$replayed", 25), false, &mut tally);
         assert_eq!(
             thread.summary.reply_count, 2,
             "a replayed or older reply was counted"
         );
-        add_reply(&mut thread, reply("$newer", 30), true, &mut counted);
-        add_reply(&mut thread, reply("$newer", 30), true, &mut counted);
+        add_reply(&mut thread, reply("$newer", 30), true, &mut tally);
+        add_reply(&mut thread, reply("$newer", 30), true, &mut tally);
         assert_eq!(thread.summary.reply_count, 3);
         assert_eq!(thread.latest_activity_ms, 30);
         assert_eq!(
@@ -985,16 +1041,40 @@ mod tests {
 
     #[test]
     fn the_latest_reply_dates_a_summary_without_one() {
-        let mut counted = HashSet::new();
+        let mut tally = Tally::default();
         let mut thread = entry("$root", None);
-        add_reply(&mut thread, reply("$latest", 50), false, &mut counted);
+        add_reply(&mut thread, reply("$latest", 50), false, &mut tally);
         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);
+        add_reply(&mut thread, reply("$latest", 50), true, &mut tally);
+        add_reply(&mut thread, reply("$replayed", 60), false, &mut tally);
         assert_eq!(thread.summary.reply_count, 2, "a replay moved the count");
     }
 
+    /// continuwuity's /threads names no latest reply, so the cache re-inserting replies the
+    /// server already counted used to count each one again
+    #[test]
+    fn replies_sent_before_the_count_was_read_never_count() {
+        let mut thread = entry("$root", None);
+        thread.summary.latest_event_id = None;
+        let mut tally = Tally::default();
+        tally.note(&Built {
+            entry: thread.clone(),
+            media: Vec::new(),
+            read_at: Some(100),
+        });
+        for (event_id, timestamp_ms) in [("$first", 40), ("$second", 60)] {
+            add_reply(&mut thread, reply(event_id, timestamp_ms), true, &mut tally);
+        }
+        assert_eq!(
+            thread.summary.reply_count, 2,
+            "already counted replies counted again"
+        );
+        assert_eq!(thread.latest_activity_ms, 60);
+        add_reply(&mut thread, reply("$after", 150), true, &mut tally);
+        assert_eq!(thread.summary.reply_count, 3);
+    }
+
     #[test]
     fn upserts_keep_newest_activity_first() {
         let mut entries = Vector::new();

Modified file: tests/state.rs+110 additions−2 deletions@@ -9,11 +9,11 @@ use macltok::{
 use matrix_sdk::{
     ThreadingSupport,
     ruma::{
-        RoomId,
+        MilliSecondsSinceUnixEpoch, RoomId,
         api::client::sync::sync_events::v5,
         event_id,
         events::{
-            AnySyncStateEvent, Mentions,
+            AnySyncStateEvent, AnyTimelineEvent, Mentions,
             receipt::{ReceiptThread, ReceiptType},
         },
         room_id,
@@ -385,3 +385,111 @@ async fn thread_unread_once(
     .await
     .expect("thread unreads never settled")
 }
+
+/// continuwuity 26.9 bundles a thread's latest reply as bare content, so nothing dates the
+/// page's count, and the cache handing back replies the server already counted counted them
+/// again
+#[tokio::test]
+async fn a_thread_list_page_keeps_the_servers_count() {
+    let server = MatrixMockServer::new().await;
+    server
+        .mock_sliding_sync()
+        .ok(v5::Response::new("1".to_owned()))
+        .mount()
+        .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!("!thread-list:localhost");
+    let own = client.user_id().unwrap().to_owned();
+    server.mock_room_state_encryption().plain().mount().await;
+    client.event_cache().subscribe().unwrap();
+    let factory = EventFactory::new().room(room_id).sender(&ALICE);
+    let (root, first, second, third) = (
+        event_id!("$list-root"),
+        event_id!("$list-first"),
+        event_id!("$list-second"),
+        event_id!("$list-third"),
+    );
+    let reply = |event_id, previous, body| {
+        factory
+            .text_msg(body)
+            .in_thread(root, previous)
+            .event_id(event_id)
+    };
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_state_event(EventFactory::new().room(room_id).member(&own))
+                .add_timeline_event(factory.text_msg("root").event_id(root))
+                .add_timeline_event(reply(first, root, "one"))
+                .add_timeline_event(reply(second, first, "two")),
+        )
+        .await;
+    let listed: Raw<AnyTimelineEvent> = serde_json::from_value(json!({
+        "type": "m.room.message",
+        "event_id": root,
+        "room_id": room_id,
+        "sender": &*ALICE,
+        "origin_server_ts": 1,
+        "content": { "msgtype": "m.text", "body": "root" },
+        "unsigned": { "m.relations": { "m.thread": {
+            "latest_event": { "msgtype": "m.text", "body": "two" },
+            "count": 2,
+            "current_user_participated": false,
+        } } },
+    }))
+    .unwrap();
+    server
+        .mock_room_threads()
+        .ok(vec![listed], None)
+        .mount()
+        .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());
+
+    handle.open_thread_list(room_id.to_owned()).await.unwrap();
+    handle
+        .paginate_thread_list(room_id.to_owned(), 20)
+        .await
+        .unwrap();
+    tokio::time::sleep(Duration::from_millis(5)).await;
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(reply(first, root, "one"))
+                .add_timeline_event(reply(second, first, "two"))
+                .add_timeline_event(
+                    reply(third, second, "three").server_ts(MilliSecondsSinceUnixEpoch::now()),
+                ),
+        )
+        .await;
+    let changed = tokio::time::timeout(
+        Duration::from_secs(10),
+        support::receive_until(&mut events, |event| {
+            matches!(event, Event::ThreadListChanged { upserts, .. }
+                if upserts.iter().any(|entry| entry.summary.latest_event_id.as_deref() == Some(third)))
+        }),
+    )
+    .await
+    .expect("the new reply never reached the thread list");
+    let Event::ThreadListChanged { upserts, .. } = changed else {
+        unreachable!()
+    };
+    let entry = upserts.last().unwrap();
+    assert_eq!(
+        entry.summary.reply_count, 3,
+        "replies the server had counted were counted again"
+    );
+
+    support::shut_down(&handle, &mut events, running).await;
+}

macltok

Commit e7c1b037154d

Commit `e7c1b037154d`