macltok

public

Star0

Fork 0

Watch0

90 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `60454002af5f`

tarball

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

7 files changed+71-51

threads: give thread list media its owner's prefix and match the close shape

The fix stack keys every media id by its owning scope's nonce so two owners of one event never share an entry, and thread list ids lacked that prefix. ThreadListClosed now carries a ThreadListCloseReason with RoomLeft, matching EventContextCloseReason, where a kick and a ban both read as leaving.

Modified file: docs/design-decisions.md+6 additions−4 deletions@@ -151,10 +151,12 @@ the latest reply's preview, and a preview's own reply target comes back
 
 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
+or banned, closes the list with `ThreadListClosed` and `RoomLeft`, the shape an
+event context closing on a leave has. `Close` sends one too, with `Requested`,
+since a mirror holds the list and has to drop it. a list's media belongs to its
+own scope and its ids carry it, like an event context's, bounded the same way so
+a long open list sheds the previews its threads moved past, and released on
+close
 
 ## warming up to send
 

Modified file: src/engine/thread_list.rs+31 additions−16 deletions@@ -21,8 +21,8 @@ use super::{
     executor::{AbortHandle, TaskSet, is_panic, resume_unwind},
 };
 use crate::{
-    Error, ErrorKind, Event, HistoryState, ThreadListClosure, ThreadListCommand, ThreadListEntry,
-    ThreadSummary, TimelineItem,
+    Error, ErrorKind, Event, HistoryState, ThreadListCloseReason, ThreadListCommand,
+    ThreadListEntry, ThreadSummary, TimelineItem,
     normalise::{
         MediaRegistration, MediaScope, member_sender, normalise_observed, resolve_replies,
     },
@@ -67,7 +67,7 @@ enum Arrival {
     Left {
         room_id: OwnedRoomId,
         generation: u64,
-        reason: ThreadListClosure,
+        reason: ThreadListCloseReason,
     },
 }
 
@@ -111,6 +111,7 @@ struct List {
     room: Room,
     generation: u64,
     scope: MediaScope,
+    media_prefix: String,
     entries: Vector<ThreadListEntry>,
     token: Option<String>,
     end_reached: bool,
@@ -184,7 +185,7 @@ impl ThreadLists {
                     return reply.done(Err(not_open()));
                 };
                 let closed = self
-                    .finish(room_id, list, ThreadListClosure::Requested)
+                    .finish(room_id, list, ThreadListCloseReason::Requested)
                     .await;
                 reply.done(closed);
             }
@@ -204,10 +205,13 @@ impl ThreadLists {
         };
         self.next_generation = self.next_generation.wrapping_add(1);
         let generation = self.next_generation;
+        let scope = self.media.context_scope();
+        let media_prefix = format!("{}#", scope.nonce());
         let listener = self.tasks.spawn(listen(
             room.clone(),
             generation,
             self.current_user_id.clone(),
+            media_prefix.clone(),
             self.arrivals.clone(),
         ));
         self.lists.insert(
@@ -215,7 +219,8 @@ impl ThreadLists {
             List {
                 room,
                 generation,
-                scope: self.media.context_scope(),
+                scope,
+                media_prefix,
                 entries: Vector::new(),
                 token: None,
                 end_reached: false,
@@ -268,6 +273,7 @@ impl ThreadLists {
             if refresh { None } else { list.token.clone() },
             list.page_limit,
             self.current_user_id.clone(),
+            list.media_prefix.clone(),
         );
         let generation = list.generation;
         let arrivals = self.arrivals.clone();
@@ -293,7 +299,7 @@ impl ThreadLists {
         &mut self,
         room_id: OwnedRoomId,
         list: List,
-        reason: ThreadListClosure,
+        reason: ThreadListCloseReason,
     ) -> Result<(), Error> {
         list.listener.abort();
         if let Load::Page(waiting) | Load::Refresh(waiting) = list.load {
@@ -486,9 +492,11 @@ impl ThreadLists {
             let room = list.room.clone();
             let generation = list.generation;
             let current_user_id = self.current_user_id.clone();
+            let media_prefix = list.media_prefix.clone();
             let arrivals = self.arrivals.clone();
             self.tasks.spawn(async move {
-                let fetched = fetch_root(&room, &root_event_id, current_user_id).await;
+                let fetched =
+                    fetch_root(&room, &root_event_id, current_user_id, &media_prefix).await;
                 let _ = arrivals.send(Arrival::Root {
                     room_id: room.room_id().to_owned(),
                     generation,
@@ -677,6 +685,7 @@ async fn fetch_page(
     from: Option<String>,
     limit: u16,
     current_user_id: Option<OwnedUserId>,
+    media_prefix: String,
 ) -> Result<Page, Error> {
     let roots = room
         .list_threads(ListThreadsOptions {
@@ -688,7 +697,9 @@ 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()).await {
+        if let Some(built) =
+            build_entry(&room, root, current_user_id.as_deref(), &media_prefix).await
+        {
             entries.push(built);
         }
     }
@@ -702,6 +713,7 @@ async fn fetch_root(
     room: &Room,
     root_event_id: &OwnedEventId,
     current_user_id: Option<OwnedUserId>,
+    media_prefix: &str,
 ) -> Fetched {
     let root = match room.load_or_fetch_event(root_event_id, None).await {
         Ok(root) => root,
@@ -719,7 +731,7 @@ async fn fetch_root(
             };
         }
     };
-    match build_entry(room, root, current_user_id.as_deref()).await {
+    match build_entry(room, root, current_user_id.as_deref(), media_prefix).await {
         Some(built) => Fetched::Built(built),
         None => Fetched::Missing,
     }
@@ -729,12 +741,13 @@ async fn build_entry(
     room: &Room,
     root: SdkTimelineEvent,
     current_user_id: Option<&ruma::UserId>,
+    media_prefix: &str,
 ) -> Option<Built> {
     let summary = root.thread_summary.summary().cloned();
     let latest = root.bundled_latest_thread_event.clone();
-    let root = normalise(room, &root, current_user_id).await?;
+    let root = normalise(room, &root, current_user_id, media_prefix).await?;
     let latest = match latest {
-        Some(latest) => normalise(room, &latest, current_user_id).await,
+        Some(latest) => normalise(room, &latest, current_user_id, media_prefix).await,
         None => None,
     };
     let summary = ThreadSummary {
@@ -769,11 +782,12 @@ async fn normalise(
     room: &Room,
     event: &SdkTimelineEvent,
     current_user_id: Option<&ruma::UserId>,
+    media_prefix: &str,
 ) -> 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 identity = format!("{media_prefix}{}\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;
@@ -795,10 +809,9 @@ async fn normalise(
     })
 }
 
-fn closure(state: RoomState) -> Option<ThreadListClosure> {
+fn closure(state: RoomState) -> Option<ThreadListCloseReason> {
     match state {
-        RoomState::Left => Some(ThreadListClosure::Left),
-        RoomState::Banned => Some(ThreadListClosure::Banned),
+        RoomState::Left | RoomState::Banned => Some(ThreadListCloseReason::RoomLeft),
         _ => None,
     }
 }
@@ -808,6 +821,7 @@ async fn listen(
     room: Room,
     generation: u64,
     current_user_id: Option<OwnedUserId>,
+    media_prefix: String,
     arrivals: mpsc::UnboundedSender<Arrival>,
 ) {
     let room_id = room.room_id().to_owned();
@@ -869,7 +883,8 @@ async fn listen(
             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(), &media_prefix).await
             else {
                 continue;
             };

Modified file: src/engine/watchers.rs+1 additions−1 deletions@@ -8,9 +8,9 @@ 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},
+    serde_helpers::extract_thread_root,
 };
 use matrix_sdk_base::RoomInfoNotableUpdateReasons;
 use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};

Modified file: src/lib.rs+14 additions−15 deletions@@ -57,23 +57,22 @@ pub use mirror::{
 pub use model::{
     AccountCommand, AttachmentEncryption, AttachmentUpload, Change, Command, CommandOutput,
     ConnectionState, CreateRoomOptions, EncryptionCommand, EncryptionState, EngineErrorKind,
-    EngineFailure, Event, EventContext, EventContextCloseReason, EventContextCommand, EventContextPage, EventContextSession,
-    EventRef, HistoryState, ImagePack, ImagePackSource, ImagePacks, KeyBackupState, LatestEvent,
-    LifecycleCommand, LinkPreview, LinkPreviewImage, Media, MediaCommand, MediaDescriptor,
-    MediaFile, MediaId, MediaKind, MediaLocation, MediaSize, MemberFilter, MemberMembership,
-    MemberSummary, MembershipChange, Message, MessageContent, MessageEditContent, MessageKind,
-    MessageMentions, NotificationCommand, OwnDevice, PackCommand, PackImage, PackImageInfo,
-    PackUsage, Patch, PowerLevelChanges, Profile, PushNotification, PushNotificationEvent,
-    PushNotificationRoom, PushNotificationSender, PushNotificationStatus, Reaction,
-    RecentHistoryState, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText,
+    EngineFailure, Event, EventContext, EventContextCloseReason, EventContextCommand,
+    EventContextPage, EventContextSession, EventRef, HistoryState, ImagePack, ImagePackSource,
+    ImagePacks, KeyBackupState, LatestEvent, LifecycleCommand, LinkPreview, LinkPreviewImage,
+    Media, MediaCommand, MediaDescriptor, MediaFile, MediaId, MediaKind, MediaLocation, MediaSize,
+    MemberFilter, MemberMembership, MemberSummary, MembershipChange, Message, MessageContent,
+    MessageEditContent, MessageKind, MessageMentions, NotificationCommand, OwnDevice, PackCommand,
+    PackImage, PackImageInfo, PackUsage, Patch, PowerLevelChanges, Profile, PushNotification,
+    PushNotificationEvent, PushNotificationRoom, PushNotificationSender, PushNotificationStatus,
+    Reaction, RecentHistoryState, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText,
     RoomCategory, RoomCommand, RoomHero, RoomListFilter, RoomMatcher, RoomMembership,
     RoomNotificationMode, RoomNotificationSettings, RoomPowerLevels, RoomPreset, RoomSummary,
-    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,
-    VerificationState, VerificationTarget,
+    RoomVisibility, SendState, Sender, SpaceChild, Termination, ThreadListCloseReason,
+    ThreadListCommand, ThreadListEntry, ThreadSummary, ThreadUnread, TimelineCommand,
+    TimelineContent, TimelineDiff, TimelineDiffError, TimelineEvent, TimelineId, TimelineItem,
+    TimelineItemId, TypingState, UploadDetails, UploadOutcome, UploadThumbnail, UtdCause,
+    VerificationAction, VerificationFailure, VerificationState, VerificationTarget,
 };
 pub use repair_model::{
     HistoryContinuity, HistoryGap, HistoryGapLifecycle, HistoryRepairCommand, HistoryRepairPhase,

Modified file: src/mirror.rs+4 additions−5 deletions@@ -21,10 +21,9 @@ use tokio::sync::broadcast;
 use crate::{
     ClientHandle, Command, ConnectionState, EncryptionState, EngineFailure, ErrorKind, Event,
     EventContextCloseReason, HistoryRepairReport, HistoryState, ImagePacks, RecentHistoryState,
-    RoomSummary, ThreadListClosure, ThreadListEntry, TimelineCommand, TimelineDiff, TimelineId,
-    TimelineItem, TypingState,
-    UploadOutcome, VerificationState, VerificationTarget, engine::executor::EngineBoxFuture,
-    transport::PublicationRegistration,
+    RoomSummary, ThreadListCloseReason, ThreadListEntry, TimelineCommand, TimelineDiff, TimelineId,
+    TimelineItem, TypingState, UploadOutcome, VerificationState, VerificationTarget,
+    engine::executor::EngineBoxFuture, transport::PublicationRegistration,
 };
 
 #[cfg(not(target_family = "wasm"))]
@@ -201,7 +200,7 @@ pub enum ThreadListChange {
         entries: Vector<ThreadListEntry>,
     },
     History(HistoryState),
-    Closed(ThreadListClosure),
+    Closed(ThreadListCloseReason),
 }
 
 /// drains the event stream into a `State` that any number of `MirrorHandle`s watch

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

Modified file: src/normalise/thread.rs+11 additions−5 deletions@@ -25,7 +25,12 @@ pub(crate) async fn stored_thread_unreads(room: &Room) -> BTreeMap<OwnedEventId,
     if room.state() != RoomState::Joined {
         return BTreeMap::new();
     }
-    match room.client().state_store().get_custom_value(&key(room)).await {
+    match room
+        .client()
+        .state_store()
+        .get_custom_value(&key(room))
+        .await
+    {
         Ok(Some(bytes)) => decode(&bytes),
         Ok(None) => BTreeMap::new(),
         Err(error) => {
@@ -51,9 +56,9 @@ pub(crate) async fn recount_thread_unreads(
         return None;
     }
     let store = match room.client().event_cache_store().lock().await {
-        Ok(MappedCrossProcessLockState::Clean(store) | MappedCrossProcessLockState::Dirty(store)) => {
-            store
-        }
+        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;
@@ -76,7 +81,8 @@ pub(crate) async fn recount_thread_unreads(
                 continue;
             }
         };
-        let mut read = unthreaded.max(read_through(room, &store, ReceiptThread::Thread(root.clone())).await);
+        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));

macltok

Commit 60454002af5f

Commit `60454002af5f`