matrix_sdk_base/sticky/
mod.rs

1// Copyright 2026 The Matrix.org Foundation C.I.C.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Sticky events ([MSC4354]), received over sliding sync through the
16//! [MSC4480] extension.
17//!
18//! A sticky event is a message-like event that the server keeps delivering to
19//! clients for a bounded duration (at most one hour), regardless of the
20//! timeline limit. Clients fold the sticky events of a room into a map keyed by
21//! `(sender, type, content.sticky_key)`, in which the event that is last to
22//! expire wins, and from which entries disappear once they expire. MatrixRTC
23//! uses this to track who is in a call.
24//!
25//! The map of a room is reachable through [`Room::sticky_events`]. It is fed
26//! from the sync response, expires its entries in the background, and
27//! broadcasts every visible change to its subscribers. Encrypted sticky events
28//! that cannot be decrypted yet are kept aside and retried when their room key
29//! arrives. Nothing is persisted: the server re-sends every sticky event that
30//! is still live when the sliding sync connection starts over.
31//!
32//! [MSC4354]: https://github.com/matrix-org/matrix-spec-proposals/pull/4354
33//! [MSC4480]: https://github.com/matrix-org/matrix-spec-proposals/pull/4480
34//! [`Room::sticky_events`]: crate::Room::sticky_events
35
36#[cfg(feature = "e2e-encryption")]
37mod decrypt;
38mod extract;
39mod map;
40mod task;
41
42#[cfg(feature = "e2e-encryption")]
43use std::collections::BTreeSet;
44use std::sync::{Arc, Mutex, OnceLock, Weak};
45
46use matrix_sdk_common::{
47    deserialized_responses::{EncryptionInfo, TimelineEventKind},
48    executor::AbortOnDrop,
49};
50use ruma::{
51    MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId,
52    events::{AnySyncTimelineEvent, TimelineEventType},
53    serde::Raw,
54};
55use tokio::sync::{Notify, broadcast};
56use tracing::{debug, warn};
57
58#[cfg(feature = "e2e-encryption")]
59pub(crate) use self::decrypt::{Decryption, decrypt, spawn_redecryptor};
60use self::map::EphemeralMap;
61pub(crate) use self::{
62    extract::{Payload, StickyMeta, classify, resolve},
63    map::Candidate,
64};
65
66/// The maximum number of encrypted sticky events kept aside per room while
67/// waiting for their room keys. Beyond that, the oldest (which are the closest
68/// to expiring) make way for new ones.
69const MAX_PENDING: usize = 5000;
70
71/// The capacity of the broadcast channel behind [`StickyEvents::subscribe`].
72const UPDATES_CHANNEL_CAPACITY: usize = 32;
73
74/// The key of a sticky event in the map of a room.
75///
76/// MSC4354 keys the map by `(room_id, sender, type, content.sticky_key)`; the
77/// room is implied by the map this key belongs to.
78#[derive(Clone, Debug, PartialEq, Eq, Hash)]
79pub struct StickyKey {
80    /// The sender of the event.
81    pub sender: OwnedUserId,
82    /// The type of the event, e.g. `m.rtc.member`.
83    pub event_type: TimelineEventType,
84    /// The `content.sticky_key` of the event.
85    pub sticky_key: String,
86}
87
88/// A sticky event that is currently live in the map of a room.
89#[derive(Clone, Debug)]
90pub struct StickyEvent {
91    /// The key of the event in the map.
92    pub key: StickyKey,
93    /// The event ID.
94    pub event_id: OwnedEventId,
95    /// The event, decrypted if it was encrypted, along with its encryption
96    /// info.
97    pub kind: TimelineEventKind,
98    /// When the event stops being sticky.
99    pub expires_at: MilliSecondsSinceUnixEpoch,
100}
101
102impl StickyEvent {
103    /// The event, decrypted if it was encrypted.
104    pub fn raw(&self) -> &Raw<AnySyncTimelineEvent> {
105        self.kind.raw()
106    }
107
108    /// The encryption info of the event, if it was encrypted.
109    pub fn encryption_info(&self) -> Option<&Arc<EncryptionInfo>> {
110        self.kind.encryption_info()
111    }
112}
113
114/// Why a sticky event disappeared from the map of a room.
115#[derive(Clone, Copy, Debug, PartialEq, Eq)]
116#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
117pub enum RemovalReason {
118    /// The event stopped being sticky.
119    Expired,
120    /// A newer event with the same key and an empty content (a removal, in
121    /// MSC4354 terms) replaced it.
122    Replaced,
123    /// We left the room.
124    RoomLeft,
125}
126
127/// A batch of visible changes to the map of a room.
128///
129/// One update is broadcast per processed sync response, and one per expiry
130/// pass, so that subscribers see the net effect of each at once.
131#[derive(Clone, Debug, Default)]
132pub struct StickyEventsUpdate {
133    /// The events that appeared under a key that had no live event.
134    pub added: Vec<StickyEvent>,
135    /// The events that replaced the live event of their key.
136    pub updated: Vec<StickyEvent>,
137    /// The keys whose live event disappeared, and why.
138    pub removed: Vec<(StickyKey, RemovalReason)>,
139}
140
141impl StickyEventsUpdate {
142    /// Whether this update carries no change at all.
143    pub fn is_empty(&self) -> bool {
144        self.added.is_empty() && self.updated.is_empty() && self.removed.is_empty()
145    }
146}
147
148/// An encrypted sticky event that couldn't be decrypted yet.
149///
150/// Without encryption support nothing ever reads the event back: it is kept
151/// until it expires, like any other, for the sake of a single code path.
152#[derive(Debug)]
153pub(crate) struct PendingEvent {
154    /// The sticky metadata of the event, with a [`Payload::Encrypted`].
155    pub meta: StickyMeta,
156    /// The encrypted event.
157    #[cfg_attr(not(feature = "e2e-encryption"), allow(dead_code))]
158    pub event: Raw<AnySyncTimelineEvent>,
159}
160
161impl PendingEvent {
162    /// Return the Megolm session the event was encrypted with (if any).
163    fn session_id(&self) -> Option<&str> {
164        match &self.meta.payload {
165            Payload::Encrypted { session_id } => session_id.as_deref(),
166            Payload::Plain { .. } => None,
167        }
168    }
169}
170
171/// The sticky events of a room.
172///
173/// This is a cheap-to-clone handle over the map of a room, which every clone
174/// (and every clone of the [`Room`](crate::Room) it belongs to) shares.
175#[derive(Clone, Debug)]
176pub struct StickyEvents {
177    inner: Arc<StickyEventsInner>,
178}
179
180#[derive(Debug)]
181struct StickyEventsInner {
182    /// The ID of the room these sticky events belong to.
183    room_id: OwnedRoomId,
184    /// The mutable state of the room's sticky events.
185    state: Mutex<State>,
186    /// Broadcasts the visible changes of the map to subscribers.
187    updates: broadcast::Sender<StickyEventsUpdate>,
188    /// Notified whenever the state changed in a way that may require the
189    /// maintenance task to wake up earlier than planned.
190    changed: Arc<Notify>,
191    /// The background task expiring entries and pending events. Spawned the
192    /// first time there is something to expire, so that a room without sticky
193    /// events costs nothing, and aborted when the last handle is dropped.
194    task: OnceLock<AbortOnDrop<()>>,
195}
196
197/// The mutable state of a room's sticky events.
198#[derive(Debug, Default)]
199struct State {
200    /// The map of live (and tombstoned) sticky events.
201    map: EphemeralMap,
202    /// Encrypted sticky events awaiting their room key, oldest first.
203    pending: Vec<PendingEvent>,
204}
205
206impl StickyEvents {
207    pub(crate) fn new(room_id: OwnedRoomId) -> Self {
208        let (updates, _) = broadcast::channel(UPDATES_CHANNEL_CAPACITY);
209
210        Self {
211            inner: Arc::new(StickyEventsInner {
212                room_id,
213                state: Default::default(),
214                updates,
215                changed: Default::default(),
216                task: OnceLock::new(),
217            }),
218        }
219    }
220
221    /// The room these sticky events belong to.
222    pub fn room_id(&self) -> &RoomId {
223        &self.inner.room_id
224    }
225
226    /// The sticky events that are currently live, in no particular order.
227    pub fn live(&self) -> Vec<StickyEvent> {
228        self.inner.state.lock().unwrap().map.live(now_ms()).collect()
229    }
230
231    /// Subscribe to the changes of the map.
232    ///
233    /// A subscriber that falls behind receives a
234    /// [`Lagged`](broadcast::error::RecvError::Lagged) error, after which it
235    /// should reconcile with [`live`](Self::live).
236    pub fn subscribe(&self) -> broadcast::Receiver<StickyEventsUpdate> {
237        self.inner.updates.subscribe()
238    }
239
240    /// Apply resolved sticky events received at `now` to the map.
241    pub(crate) fn ingest(&self, now: u64, candidates: Vec<Candidate>) {
242        if candidates.is_empty() {
243            return;
244        }
245
246        self.ensure_task();
247
248        let update = self.inner.state.lock().unwrap().map.apply(now, candidates);
249        self.inner.changed.notify_one();
250        self.publish(update);
251    }
252
253    /// Keep encrypted sticky events aside until their room key arrives.
254    pub(crate) fn park(&self, pending: Vec<PendingEvent>) {
255        if pending.is_empty() {
256            return;
257        }
258
259        self.ensure_task();
260
261        {
262            let mut state = self.inner.state.lock().unwrap();
263
264            for event in pending {
265                if event.session_id().is_none() {
266                    debug!(
267                        room_id = %self.inner.room_id,
268                        event_id = %event.meta.event_id,
269                        "Dropping an encrypted sticky event without a session ID"
270                    );
271                    continue;
272                }
273
274                state.pending.push(event);
275            }
276
277            if state.pending.len() > MAX_PENDING {
278                warn!(
279                    room_id = %self.inner.room_id,
280                    "Too many encrypted sticky events await their room key, dropping the oldest"
281                );
282                let excess = state.pending.len() - MAX_PENDING;
283                state.pending.drain(..excess);
284            }
285        }
286
287        self.inner.changed.notify_one();
288    }
289
290    /// Take the pending encrypted sticky events that were encrypted with one of
291    /// `session_ids`, or all of them if `None`.
292    #[cfg(feature = "e2e-encryption")]
293    pub(crate) fn take_pending(&self, session_ids: Option<&BTreeSet<String>>) -> Vec<PendingEvent> {
294        let mut state = self.inner.state.lock().unwrap();
295
296        match session_ids {
297            None => std::mem::take(&mut state.pending),
298            Some(session_ids) => {
299                let (taken, kept) =
300                    std::mem::take(&mut state.pending).into_iter().partition(|event| {
301                        event.session_id().is_some_and(|id| session_ids.contains(id))
302                    });
303                state.pending = kept;
304                taken
305            }
306        }
307    }
308
309    /// Whether any encrypted sticky event awaits its room key.
310    #[cfg(feature = "e2e-encryption")]
311    pub(crate) fn has_pending(&self) -> bool {
312        !self.inner.state.lock().unwrap().pending.is_empty()
313    }
314
315    /// Forget every sticky event, e.g. because we left the room.
316    pub(crate) fn clear(&self) {
317        let removed = {
318            let mut state = self.inner.state.lock().unwrap();
319            state.pending.clear();
320            state.map.clear(now_ms())
321        };
322
323        self.publish(StickyEventsUpdate {
324            removed: removed.into_iter().map(|key| (key, RemovalReason::RoomLeft)).collect(),
325            ..Default::default()
326        });
327    }
328
329    fn publish(&self, update: StickyEventsUpdate) {
330        if !update.is_empty() {
331            // Failing to send only means there is no subscriber.
332            let _ = self.inner.updates.send(update);
333        }
334    }
335
336    fn ensure_task(&self) {
337        self.inner
338            .task
339            .get_or_init(|| task::spawn(Arc::downgrade(&self.inner), self.inner.changed.clone()));
340    }
341}
342
343impl StickyEventsInner {
344    /// Drop what has expired at `now`, broadcast the visible removals, and
345    /// return when the next entry is due to expire, if any.
346    fn expire(&self, now: u64) -> Option<u64> {
347        let (removed, next) = {
348            let mut state = self.state.lock().unwrap();
349
350            let removed = state.map.evict_expired(now);
351            state.pending.retain(|event| event.meta.expires_at > now);
352
353            let next_pending = state.pending.iter().map(|event| event.meta.expires_at).min();
354            let next = match (state.map.next_expiry(), next_pending) {
355                (Some(a), Some(b)) => Some(a.min(b)),
356                (a, b) => a.or(b),
357            };
358
359            (removed, next)
360        };
361
362        if !removed.is_empty() {
363            let _ = self.updates.send(StickyEventsUpdate {
364                removed: removed.into_iter().map(|key| (key, RemovalReason::Expired)).collect(),
365                ..Default::default()
366            });
367        }
368
369        next
370    }
371}
372
373/// The current time, in milliseconds since the Unix epoch.
374pub(crate) fn now_ms() -> u64 {
375    MilliSecondsSinceUnixEpoch::now().get().into()
376}
377
378/// A weak handle to the state of a room's sticky events, for background
379/// tasks.
380type WeakStickyEvents = Weak<StickyEventsInner>;
381
382#[cfg(test)]
383mod tests {
384    use std::time::Duration;
385
386    use matrix_sdk_common::deserialized_responses::TimelineEventKind;
387    use matrix_sdk_test::async_test;
388    use ruma::{
389        events::{AnySyncTimelineEvent, TimelineEventType},
390        owned_event_id, owned_user_id, room_id,
391        serde::Raw,
392    };
393    use serde_json::json;
394    use strass::assert_let;
395    use tokio::sync::broadcast::error::TryRecvError;
396
397    use super::{
398        Candidate, PendingEvent, RemovalReason, StickyEvents, StickyKey, StickyMeta, now_ms,
399    };
400    use crate::sticky::Payload;
401
402    fn raw_event() -> Raw<AnySyncTimelineEvent> {
403        serde_json::from_value(json!({
404            "type": "m.rtc.member",
405            "sender": "@alice:localhost",
406            "event_id": "$a:localhost",
407            "origin_server_ts": 1,
408            "content": { "msc4354_sticky_key": "slot" },
409        }))
410        .unwrap()
411    }
412
413    fn candidate(sticky_key: &str, expires_at: u64) -> Candidate {
414        Candidate {
415            key: StickyKey {
416                sender: owned_user_id!("@alice:localhost"),
417                event_type: TimelineEventType::from("m.rtc.member"),
418                sticky_key: sticky_key.to_owned(),
419            },
420            event_id: owned_event_id!("$a:localhost"),
421            order_ts: expires_at,
422            expires_at,
423            is_tombstone: false,
424            kind: TimelineEventKind::PlainText { event: raw_event() },
425        }
426    }
427
428    fn pending(session_id: Option<&str>, expires_at: u64) -> PendingEvent {
429        PendingEvent {
430            meta: StickyMeta {
431                sender: owned_user_id!("@alice:localhost"),
432                event_id: owned_event_id!("$a:localhost"),
433                order_ts: expires_at,
434                expires_at,
435                payload: Payload::Encrypted { session_id: session_id.map(ToOwned::to_owned) },
436            },
437            event: raw_event(),
438        }
439    }
440
441    #[async_test]
442    async fn test_ingest_broadcasts_and_exposes_live_events() {
443        let sticky = StickyEvents::new(room_id!("!room:localhost").to_owned());
444        let mut subscriber = sticky.subscribe();
445
446        let now = now_ms();
447        sticky.ingest(now, vec![candidate("slot", now + 60_000)]);
448
449        let live = sticky.live();
450        assert_eq!(live.len(), 1);
451        assert_eq!(live[0].key.sticky_key, "slot");
452        assert!(live[0].encryption_info().is_none());
453
454        let update = subscriber.try_recv().unwrap();
455        assert_eq!(update.added.len(), 1);
456        assert_eq!(update.added[0].key.sticky_key, "slot");
457    }
458
459    #[async_test]
460    async fn test_expired_events_are_evicted_in_the_background() {
461        let sticky = StickyEvents::new(room_id!("!room:localhost").to_owned());
462        let mut subscriber = sticky.subscribe();
463
464        let now = now_ms();
465        sticky.ingest(now, vec![candidate("slot", now + 50)]);
466        assert_let!(Ok(update) = subscriber.try_recv());
467        assert_eq!(update.added.len(), 1);
468
469        // No further action on our side: the maintenance task must notice the
470        // expiry and tell us.
471        let update = tokio::time::timeout(Duration::from_secs(5), subscriber.recv())
472            .await
473            .expect("the expiry should be broadcast in time")
474            .unwrap();
475        assert_let!([(key, RemovalReason::Expired)] = update.removed.as_slice());
476        assert_eq!(key.sticky_key, "slot");
477
478        assert!(sticky.live().is_empty());
479    }
480
481    #[async_test]
482    async fn test_clear_broadcasts_removals() {
483        let sticky = StickyEvents::new(room_id!("!room:localhost").to_owned());
484        let mut subscriber = sticky.subscribe();
485
486        let now = now_ms();
487        sticky.ingest(now, vec![candidate("slot", now + 60_000)]);
488        sticky.park(vec![pending(Some("session"), now + 60_000)]);
489        let _ = subscriber.try_recv().unwrap();
490
491        sticky.clear();
492
493        let update = subscriber.try_recv().unwrap();
494        assert_let!([(key, RemovalReason::RoomLeft)] = update.removed.as_slice());
495        assert_eq!(key.sticky_key, "slot");
496        assert!(sticky.live().is_empty());
497        assert_let!(Err(TryRecvError::Empty) = subscriber.try_recv());
498    }
499
500    #[cfg(feature = "e2e-encryption")]
501    #[async_test]
502    async fn test_pending_events_are_taken_by_session_id() {
503        use std::collections::BTreeSet;
504
505        let sticky = StickyEvents::new(room_id!("!room:localhost").to_owned());
506
507        let now = now_ms();
508        sticky.park(vec![
509            pending(Some("session1"), now + 60_000),
510            pending(Some("session2"), now + 60_000),
511            // Nothing could ever decrypt this one: it isn't kept.
512            pending(None, now + 60_000),
513        ]);
514
515        let taken = sticky.take_pending(Some(&BTreeSet::from(["session2".to_owned()])));
516        assert_eq!(taken.len(), 1);
517        assert_let!(Payload::Encrypted { session_id } = &taken[0].meta.payload);
518        assert_eq!(session_id.as_deref(), Some("session2"));
519
520        assert!(sticky.has_pending());
521        let taken = sticky.take_pending(None);
522        assert_eq!(taken.len(), 1);
523        assert!(!sticky.has_pending());
524    }
525
526    #[cfg(feature = "e2e-encryption")]
527    #[async_test]
528    async fn test_pending_events_are_capped() {
529        use std::collections::BTreeSet;
530
531        use super::MAX_PENDING;
532
533        let sticky = StickyEvents::new(room_id!("!room:localhost").to_owned());
534
535        let now = now_ms();
536        // One more than the cap, the oldest first.
537        let mut events = vec![pending(Some("oldest"), now + 60_000)];
538        events.extend((0..MAX_PENDING).map(|_| pending(Some("newer"), now + 60_000)));
539        sticky.park(events);
540
541        // The oldest made way.
542        assert!(sticky.take_pending(Some(&BTreeSet::from(["oldest".to_owned()]))).is_empty());
543        assert_eq!(sticky.take_pending(None).len(), MAX_PENDING);
544    }
545
546    #[cfg(feature = "e2e-encryption")]
547    #[async_test]
548    async fn test_expired_pending_events_are_pruned_in_the_background() {
549        let sticky = StickyEvents::new(room_id!("!room:localhost").to_owned());
550
551        sticky.park(vec![pending(Some("session"), now_ms() + 50)]);
552        assert!(sticky.has_pending());
553
554        tokio::time::timeout(Duration::from_secs(5), async {
555            while sticky.has_pending() {
556                tokio::time::sleep(Duration::from_millis(20)).await;
557            }
558        })
559        .await
560        .expect("the pending event should be pruned in time");
561    }
562}
matrix_sdk_base/sticky/mod.rs

matrix_sdk_base/sticky/
mod.rs
