macltok

public

Star0

Fork 0

Watch0

100 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `6b04099c763c`

tarball

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

5 files changed+162-69

mentions: never wait on the event cache or the state store while observing

The offline reply and profile lookups awaited the event cache and the state store inside the mentions actor, which the sync commit hook waits on, and a resumed hosted inbox stopped answering mentions_page. Observing now settles a reply only from events the inbox already holds, and profiles and the rest come from the capture, which runs beside the actor.

Modified file: src/engine/context.rs+1 additions−47 deletions@@ -18,7 +18,7 @@ use crate::{
     engine::MediaService,
     normalise::{
         MediaRegistration, MediaScope, apply_profiles, member_sender, normalise_context,
-        normalise_observed, resolve_replies, stored_member_sender,
+        normalise_observed, resolve_replies,
     },
     transport::EventSender,
 };
@@ -550,52 +550,6 @@ async fn fetch_reply(
     }
 }
 
-/// settles an event built outside any page from what the client already holds, so it never
-/// waits on the network. the sender gets its stored profile and the reply comes from `known`
-/// or the event cache, or is unavailable
-pub(crate) async fn settle_offline(
-    room: &Room,
-    current_user_id: Option<&ruma::UserId>,
-    item: &mut TimelineItem,
-    known: impl Fn(&ruma::EventId) -> Option<TimelineEvent>,
-) {
-    let TimelineItem::Event(event) = item else {
-        return;
-    };
-    if let Some(sender) = stored_member_sender(room, &event.sender.user_id).await {
-        event.sender = sender;
-    }
-    let Some(reply) = &event.reply_to else {
-        return;
-    };
-    let target_id = reply.event_id.clone();
-    let mut target = known(&target_id);
-    if target.is_none()
-        && let Ok((cache, _handles)) = room.event_cache().await
-        && let Ok(Some(cached)) = cache.find_event(&target_id).await
-    {
-        target = serde_json::from_str::<Value>(cached.raw().json().get())
-            .ok()
-            .and_then(|value| {
-                let identity = format!("{}\n{target_id}", room.room_id());
-                match normalise_observed(&value, current_user_id, &identity)? {
-                    (TimelineItem::Event(event), _) => Some(*event),
-                    _ => None,
-                }
-            });
-    }
-    if let Some(target) = &mut target
-        && let Some(sender) = stored_member_sender(room, &target.sender.user_id).await
-    {
-        target.sender = sender;
-    }
-    let targets: HashMap<_, _> = target
-        .map(|target| (target_id, target))
-        .into_iter()
-        .collect();
-    resolve_replies(std::iter::once(&mut **event), &targets);
-}
-
 /// under lazy-loaded members a sender may be unknown locally, and `get_member` fetches the
 /// member list once for the room when it is
 async fn profiles<'a>(

Modified file: src/mentions/service.rs+16 additions−12 deletions@@ -1,5 +1,5 @@
 use std::{
-    collections::{BTreeMap, BTreeSet},
+    collections::{BTreeMap, BTreeSet, HashMap},
     sync::{
         Arc,
         atomic::{AtomicBool, Ordering},
@@ -32,7 +32,7 @@ use crate::{
     MentionGapReason, MentionKey, MentionReason, MentionsCommand, MentionsPage, MentionsUpdate,
     TimelineContent, TimelineItem,
     engine::{MediaService, MentionsStore, context, executor::EngineBoxFuture},
-    normalise::{MediaRegistration, MediaScope, normalise_observed},
+    normalise::{MediaRegistration, MediaScope, normalise_observed, resolve_replies},
     transport::EventSender,
 };
 
@@ -897,18 +897,22 @@ impl Service {
         else {
             return;
         };
-        if let Some(room) = self.client.get_room(room_id) {
-            let known = |event_id: &ruma::EventId| {
-                let key = id(&MentionKey {
-                    room_id: room_id.clone(),
-                    event_id: event_id.to_owned(),
-                });
-                match &data.events.get(&key)?.item {
-                    TimelineItem::Event(event) => Some((**event).clone()),
-                    _ => None,
+        // the commit hook waits on this actor, so a reply settles here only from what the
+        // inbox already holds, and the capture fetches profiles and anything else later
+        if let TimelineItem::Event(event) = &mut item
+            && let Some(reply) = &event.reply_to
+        {
+            let target = id(&MentionKey {
+                room_id: room_id.clone(),
+                event_id: reply.event_id.clone(),
+            });
+            let targets: HashMap<_, _> = match data.events.get(&target).map(|saved| &saved.item) {
+                Some(TimelineItem::Event(saved)) => {
+                    HashMap::from([(reply.event_id.clone(), (**saved).clone())])
                 }
+                _ => HashMap::new(),
             };
-            context::settle_offline(&room, Some(&self.account), &mut item, known).await;
+            resolve_replies(std::iter::once(&mut **event), &targets);
         }
         self.supplement(data, &key, &item, &media, &value);
         let (reasons, suppressed, unresolved) = self.classify(data, room_id, &value).await;

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

@@ -33,7 +33,7 @@ pub(crate) async fn member_sender(room: &Room, user_id: &OwnedUserId) -> Option<

3333 }

3434

3535 /// what the state store already knows, without asking the server

36-pub(crate) async fn stored_member_sender(room: &Room, user_id: &OwnedUserId) -> Option<Sender> {

36+async fn stored_member_sender(room: &Room, user_id: &OwnedUserId) -> Option<Sender> {

3737 let member = room.get_member_no_sync(user_id).await.ok().flatten()?;

3838 Some(sender(user_id, &member))

3939 }

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

@@ -15,7 +15,7 @@ pub(crate) use image_pack::{

1515 };

1616 pub(crate) use link_preview::{ROOM_PREVIEW_URLS, server_preview, url_previews_allowed};

1717 pub(crate) use media::{MediaRegistration, MediaScope, MediaSourceInfo};

18-pub(crate) use member::{member_sender, member_summary, stored_member_sender};

18+pub(crate) use member::{member_sender, member_summary};

1919 pub(crate) use observed::{normalise_observed, poll_question};

2020 pub(crate) use room::notification_mode;

2121 pub(crate) use room::{room_summaries, space_child, stored_space_order};

Modified file: tests/mentions_inbox.rs+143 additions−8 deletions@@ -1154,14 +1154,43 @@ async fn inbox_rows_carry_sender_profiles_and_never_leave_replies_pending() {
         event["content"]["m.relates_to"] = json!({ "m.in_reply_to": { "event_id": target } });
         event
     };
-    feed.replace(
-        room_id,
-        vec![
-            reply("$answer", "$parent", 2_000),
-            reply("$orphan", "$missing", 2_100),
-        ],
-    );
-    let page = mentions_page(&mut events, room_id, |page| page.entries.len() == 2).await;
+    let answer = reply("$answer", "$parent", 2_000);
+    let orphan = reply("$orphan", "$missing", 2_100);
+    for event in [&answer, &orphan] {
+        let mut event = event.clone();
+        event["room_id"] = json!(room_id);
+        server
+            .mock_room_event_context()
+            .room(room_id.to_owned())
+            .match_event_id()
+            .ok(RoomContextResponseTemplate::new(
+                matrix_sdk::deserialized_responses::TimelineEvent::from_plaintext(
+                    ruma::serde::Raw::new(&event).unwrap().cast_unchecked(),
+                ),
+            ))
+            .mount()
+            .await;
+    }
+    feed.replace(room_id, vec![answer, orphan]);
+    let settled = |page: &MentionsPage| {
+        page.entries.len() == 2
+            && page.entries.iter().all(|entry| {
+                matches!(&entry.event, macltok::TimelineItem::Event(event)
+                if event.sender.display_name.is_some()
+                    && event.reply_to.as_ref().is_some_and(|reply| {
+                        !matches!(reply.details, macltok::ReplyDetails::Pending)
+                    }))
+            })
+    };
+    let page = mentions_page(&mut events, room_id, |page| {
+        settled(page)
+            && page.entries.iter().any(|entry| {
+                matches!(&entry.event, macltok::TimelineItem::Event(event)
+                    if matches!(event.reply_to.as_ref().unwrap().details,
+                        macltok::ReplyDetails::Ready { .. }))
+            })
+    })
+    .await;
     for entry in &page.entries {
         let macltok::TimelineItem::Event(event) = &entry.event else {
             panic!("a mention should be an event");
@@ -1179,3 +1208,109 @@ async fn inbox_rows_carry_sender_profiles_and_never_leave_replies_pending() {
     }
     shut_down(&handle, &mut events, running).await;
 }
+
+#[cfg(feature = "sqlite")]
+#[tokio::test]
+async fn a_resumed_inbox_with_threaded_replies_still_answers_a_page() {
+    static NEXT_TEMPORARY_PATH: AtomicUsize = AtomicUsize::new(0);
+
+    let server = MatrixMockServer::new().await;
+    server
+        .mock_versions()
+        .with_simplified_sliding_sync()
+        .ok()
+        .mount()
+        .await;
+    server.mock_login().ok().mount().await;
+    let root = std::env::temp_dir().join(format!(
+        "macltok-mentions-threads-{}-{}",
+        std::process::id(),
+        NEXT_TEMPORARY_PATH.fetch_add(1, Ordering::Relaxed)
+    ));
+    let sessions = Arc::new(EncryptedSessionStore::file(
+        root.join("records").join("session"),
+        StoreConfig::sqlite(root.join("store"), "mentions thread test passphrase"),
+    ));
+    let room_id = room_id!("!mentions-threads:matrix.org");
+    let own_user = user_id!("@cheeky_monkey:matrix.org");
+    let alice = "@alice:matrix.org";
+    let threaded = |event_id: &str, timestamp| {
+        let mut event = message(
+            event_id,
+            alice,
+            timestamp,
+            "threaded mention",
+            Some(own_user.as_str()),
+        );
+        event["content"]["m.relates_to"] = json!({
+            "rel_type": "m.thread",
+            "event_id": "$root",
+            "is_falling_back": true,
+            "m.in_reply_to": { "event_id": "$root" },
+        });
+        event
+    };
+    let feed = SlidingFeed::new(
+        room_id.to_owned(),
+        vec![
+            message("$root", alice, 6_000, "thread root", None),
+            threaded("$thread-a", 7_000),
+            threaded("$thread-b", 7_100),
+            message(
+                "$plain",
+                alice,
+                7_200,
+                "plain mention",
+                Some(own_user.as_str()),
+            ),
+        ],
+    );
+    feed.set_required_state(
+        room_id,
+        vec![json!({
+            "type": "m.room.member",
+            "event_id": "$joined-member",
+            "sender": own_user,
+            "state_key": own_user,
+            "origin_server_ts": 1,
+            "content": { "membership": "join" },
+        })],
+    );
+    feed.mount(&server).await;
+
+    let config = EngineConfig::default()
+        .with_initial_history_limit(0)
+        .with_search_backfill_pages_per_room(0);
+    let session = Session::login_password(
+        PasswordLogin::new(server.uri(), "example", "wordpass").allow_insecure_loopback(true),
+        sessions.clone(),
+    )
+    .await
+    .unwrap();
+    let (engine, handle, mut events) = session.build(config.clone()).await.unwrap();
+    let running = tokio::spawn(engine.run());
+    mentions_page(&mut events, room_id, |page| page.entries.len() == 3).await;
+    shut_down(&handle, &mut events, running).await;
+    drop(handle);
+
+    let Some(Restore::Resumed(session)) = Session::restore(sessions.clone()).await.unwrap() else {
+        panic!("the saved session did not resume");
+    };
+    let (engine, handle, mut events) = session.build(config).await.unwrap();
+    let running = tokio::spawn(engine.run());
+    receive_until(&mut events, |event| {
+        matches!(event, Event::Authenticated { .. })
+    })
+    .await;
+    wait_for_requests(&feed, 4).await;
+    let resumed = tokio::time::timeout(Duration::from_secs(5), handle.mentions_page(None, 50))
+        .await
+        .expect("the resumed inbox did not answer a page")
+        .unwrap();
+    assert_eq!(resumed.entries.len(), 3);
+    shut_down(&handle, &mut events, running).await;
+    drop(handle);
+
+    sessions.discard().await.unwrap();
+    std::fs::remove_dir_all(root).unwrap();
+}

macltok

Commit 6b04099c763c

Commit `6b04099c763c`