macltok

public

Star0

Fork 0

Watch0

121 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `3eee54c8ed63`

tarball

atagen <boss@atagen.co>2026-10-27 03:38 UTC

3 files changed+232-9

mentions: give saved rows a sender profile after their capture

observe normalises live events with no profile, since it must not wait on the state store under the commit hook (6b04099), and supplement adds them to windows whose capture already finished, so they kept bare user ids for good. A pass outside the actor now looks bare senders up per room through the capture's profiles(), once per sender a run. profiles() also marks a room's members missing once when a sender isn't found, since stores have kept members_synced over a partial roster.

Modified file: src/engine/context.rs+27 additions−9 deletions@@ -552,22 +552,40 @@ async fn fetch_reply(
 
 /// 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>(
+pub(crate) async fn profiles<'a>(
     room: &Room,
     user_ids: impl Iterator<Item = &'a OwnedUserId>,
 ) -> HashMap<OwnedUserId, Sender> {
     let mut unique: Vec<&OwnedUserId> = user_ids.collect();
     unique.sort();
     unique.dedup();
-    join_all(
-        unique.into_iter().map(|user_id| async move {
+    let lookup = |users: Vec<&'a OwnedUserId>| {
+        join_all(users.into_iter().map(|user_id| async move {
             Some((user_id.clone(), member_sender(room, user_id).await?))
-        }),
-    )
-    .await
-    .into_iter()
-    .flatten()
-    .collect()
+        }))
+    };
+    let mut found: HashMap<_, _> = lookup(unique.clone()).await.into_iter().flatten().collect();
+    let missing: Vec<_> = unique
+        .into_iter()
+        .filter(|user| !found.contains_key(*user))
+        .collect();
+    // stores have been seen with members_synced persisted over a partial roster, after which
+    // the sdk never asks /members again, so a sender it can't find marks it stale once a run
+    if !missing.is_empty() && members_rechecked(room) {
+        room.mark_members_missing();
+        found.extend(lookup(missing).await.into_iter().flatten());
+    }
+    found
+}
+
+fn members_rechecked(room: &Room) -> bool {
+    static CHECKED: std::sync::LazyLock<
+        std::sync::Mutex<std::collections::HashSet<(Option<OwnedUserId>, ruma::OwnedRoomId)>>,
+    > = std::sync::LazyLock::new(Default::default);
+    CHECKED.lock().unwrap().insert((
+        room.client().user_id().map(ToOwned::to_owned),
+        room.room_id().to_owned(),
+    ))
 }
 
 fn validate_limit(limit: u16) -> Result<(), Error> {

Modified file: src/mentions/service.rs+120 additions@@ -118,6 +118,10 @@ pub(crate) struct Service {
     recovery_budget: usize,
     latest_scan: bool,
     recovery_retry: u64,
+    /// senders a profile pass already looked up, found or not, so one without a profile isn't
+    /// asked for again every second
+    profiled: BTreeSet<(OwnedRoomId, OwnedUserId)>,
+    profiling: bool,
 }
 
 enum Work {
@@ -132,6 +136,11 @@ enum Work {
         Option<String>,
         Result<ruma::api::client::push::get_notifications::v3::Response, Error>,
     ),
+    Profiles(
+        OwnedRoomId,
+        Vec<OwnedUserId>,
+        HashMap<OwnedUserId, crate::Sender>,
+    ),
 }
 
 impl Work {
@@ -148,6 +157,7 @@ impl Work {
                 .map(|notification| notification.room_id.clone())
                 .collect(),
             Self::History(..) => BTreeSet::new(),
+            Self::Profiles(room_id, ..) => BTreeSet::from([room_id.clone()]),
         }
     }
 }
@@ -326,6 +336,8 @@ impl Service {
             recovery_budget: 20,
             latest_scan: true,
             recovery_retry: 0,
+            profiled: BTreeSet::new(),
+            profiling: false,
         };
         service.ignored = service.ignored_users().await;
         Ok((
@@ -1208,6 +1220,7 @@ impl Service {
                 }
                 () = &mut tick => {
                     tick = Box::pin(matrix_sdk::sleep::sleep(Duration::from_secs(1)));
+                    self.schedule_profiles(&mut work);
                     if now() >= next_scan && !history { self.recovery_budget = 20; self.latest_scan = true; next_scan = now() + 60_000; }
                     let direct = direct_rooms(&self.client);
                     if direct != self.direct { self.direct = direct; self.dirty = true; self.unpublished = true; }
@@ -1306,6 +1319,47 @@ impl Service {
                     }
                 }
             }
+            Work::Profiles(room_id, users, found) => {
+                self.profiling = false;
+                self.profiled
+                    .extend(users.into_iter().map(|user| (room_id.clone(), user)));
+                let bare = |item: &TimelineItem| match item {
+                    TimelineItem::Event(event)
+                        if event.sender.display_name.is_none() && event.sender.avatar.is_none() =>
+                    {
+                        found.get(&event.sender.user_id).cloned()
+                    }
+                    _ => None,
+                };
+                let prefix = format!("{room_id}\n");
+                let mut repaired = Vec::new();
+                for (key, saved) in next.events.range_mut(prefix.clone()..) {
+                    if !key.starts_with(&prefix) {
+                        break;
+                    }
+                    if let Some(sender) = bare(&saved.item)
+                        && let TimelineItem::Event(event) = &mut saved.item
+                    {
+                        event.sender = sender;
+                        repaired.push(key.clone());
+                    }
+                }
+                let revision = next.revision + 1;
+                for (key, entry) in next.entries.range_mut(prefix.clone()..) {
+                    if !key.starts_with(&prefix) {
+                        break;
+                    }
+                    if let Some(sender) = bare(&entry.event)
+                        && let TimelineItem::Event(event) = &mut entry.event
+                    {
+                        event.sender = sender;
+                        entry.revision = revision;
+                    }
+                }
+                for key in &repaired {
+                    next.touch_contexts(key);
+                }
+            }
             Work::History(latest, from, result) => {
                 *history = false;
                 self.recovery_budget = self.recovery_budget.saturating_sub(1);
@@ -1387,6 +1441,72 @@ impl Service {
         true
     }
 
+    /// rows observed after their window's capture never got a profile, so one room at a time
+    /// looks its bare senders up the way a capture does, outside the actor
+    fn schedule_profiles(&mut self, work: &mut FuturesUnordered<EngineBoxFuture<'static, Work>>) {
+        if self.profiling {
+            return;
+        }
+        let mut wanted: Option<(OwnedRoomId, BTreeSet<OwnedUserId>)> = None;
+        let items = self
+            .data
+            .events
+            .iter()
+            .map(|(key, saved)| (key, &saved.item))
+            .chain(
+                self.data
+                    .entries
+                    .iter()
+                    .map(|(key, entry)| (key, &entry.event)),
+            );
+        for (key, item) in items {
+            let TimelineItem::Event(event) = item else {
+                continue;
+            };
+            if event.sender.display_name.is_some() || event.sender.avatar.is_some() {
+                continue;
+            }
+            let Some(room_id) = key
+                .split_once('\n')
+                .and_then(|(room, _)| OwnedRoomId::try_from(room).ok())
+            else {
+                continue;
+            };
+            if self.direct.contains(&room_id)
+                || self
+                    .profiled
+                    .contains(&(room_id.clone(), event.sender.user_id.clone()))
+            {
+                continue;
+            }
+            match &mut wanted {
+                None => wanted = Some((room_id, BTreeSet::from([event.sender.user_id.clone()]))),
+                Some((room, users)) if *room == room_id && users.len() < 50 => {
+                    users.insert(event.sender.user_id.clone());
+                }
+                Some(_) => {}
+            }
+        }
+        let Some((room_id, users)) = wanted else {
+            return;
+        };
+        let Some(room) = self
+            .client
+            .get_room(&room_id)
+            .filter(|room| room.state() == RoomState::Joined)
+        else {
+            self.profiled
+                .extend(users.into_iter().map(|user| (room_id.clone(), user)));
+            return;
+        };
+        self.profiling = true;
+        let users: Vec<_> = users.into_iter().collect();
+        work.push(Box::pin(async move {
+            let found = context::profiles(&room, users.iter()).await;
+            Work::Profiles(room_id, users, found)
+        }));
+    }
+
     fn schedule(
         &mut self,
         work: &mut FuturesUnordered<EngineBoxFuture<'static, Work>>,

Modified file: src/mentions/service_tests.rs+85 additions@@ -778,3 +778,88 @@ async fn observing_again_keeps_the_profiles_a_capture_fetched() {
     );
     assert_eq!(data.entries[&ping].capture, capture);
 }
+
+#[tokio::test]
+async fn a_profile_pass_names_rows_observed_after_their_capture() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!mentions:localhost");
+    let (client, own_user) = joined_client(&server, room_id).await;
+    let bob = user_id!("@bob:localhost");
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id).add_state_event(
+                EventFactory::new()
+                    .room(room_id)
+                    .member(bob)
+                    .display_name("Bob"),
+            ),
+        )
+        .await;
+    let raw = |event: matrix_sdk::ruma::serde::Raw<_>, timestamp: u64| {
+        let mut value: serde_json::Value = serde_json::from_str(event.json().get()).unwrap();
+        value["origin_server_ts"] = json!(timestamp);
+        value
+    };
+    let factory = EventFactory::new().room(room_id).sender(bob);
+    let mention = raw(
+        factory
+            .text_msg("ping")
+            .mentions(Mentions::with_user_ids([own_user]))
+            .event_id(event_id!("$ping:localhost"))
+            .into_raw_sync(),
+        10,
+    );
+    let following = raw(
+        factory
+            .text_msg("later")
+            .event_id(event_id!("$later:localhost"))
+            .into_raw_sync(),
+        11,
+    );
+    let mut service = service(&client, None).await;
+    let room_id = room_id.to_owned();
+    let mut data = Data::default();
+    service.observe(&mut data, &room_id, mention).await;
+    let ping = format!("{room_id}\n$ping:localhost");
+    let revision = data.entries[&ping].revision;
+    let target = data.entries[&ping].event.clone();
+    service.captured(
+        &mut data,
+        ping.clone(),
+        revision,
+        Ok((
+            crate::EventContext {
+                events_before: Vec::new(),
+                event: target,
+                events_after: Vec::new(),
+                start_token: None,
+                end_token: None,
+            },
+            Vec::new(),
+            Vec::new(),
+        )),
+    );
+    service.observe(&mut data, &room_id, following).await;
+    let later = format!("{room_id}\n$later:localhost");
+    let name = |item: &TimelineItem| match item {
+        TimelineItem::Event(event) => event.sender.display_name.clone(),
+        _ => None,
+    };
+    assert_eq!(name(&data.events[&later].item), None);
+
+    service.data = data;
+    let mut work = FuturesUnordered::new();
+    service.schedule_profiles(&mut work);
+    let result = work.next().await.expect("a profile pass was scheduled");
+    let mut next = std::mem::take(&mut service.data);
+    let (mut captures, mut retries, mut history) = (BTreeSet::new(), BTreeSet::new(), false);
+    service
+        .complete(&mut next, result, &mut captures, &mut retries, &mut history)
+        .await;
+    assert_eq!(name(&next.events[&later].item).as_deref(), Some("Bob"));
+    assert_eq!(name(&next.entries[&ping].event).as_deref(), Some("Bob"));
+    service.data = next;
+    service.schedule_profiles(&mut work);
+    assert!(work.is_empty());
+}

macltok

Commit 3eee54c8ed63

Commit `3eee54c8ed63`