matrix-rust-sdk

public

Star0

Fork 0

Watch0

48 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `f05c136873e9`

tarball

atagen <boss@atagen.co>2026-10-27 04:45 UTC

5 files changed+61-5

sliding-sync: save pos only after the event cache handled the response

`RoomUpdates` reach the event cache through a broadcast, so `cache_to_storage` could persist a `pos` before the response's events were stored, and a kill in between resumed past them with no gap to back-paginate. The rooms connection now waits for `room_updates_task` to catch up with the client's send count, for at most 10s, after which pos is saved anyway as before.

Modified file: crates/matrix-sdk/src/client/mod.rs+18 additions−1 deletions@@ -19,7 +19,10 @@ use std::{
     fmt::{self, Debug},
     future::{Future, ready},
     pin::Pin,
-    sync::{Arc, Mutex as StdMutex, RwLock as StdRwLock, Weak},
+    sync::{
+        Arc, Mutex as StdMutex, RwLock as StdRwLock, Weak,
+        atomic::{AtomicU64, Ordering},
+    },
     time::Duration,
 };
 
@@ -348,6 +351,9 @@ pub(crate) struct ClientInner {
     /// sync response.
     pub(crate) room_updates_sender: broadcast::Sender<RoomUpdates>,
 
+    /// How many [`RoomUpdates`] have been sent on `room_updates_sender`.
+    room_updates_sent: AtomicU64,
+
     /// Whether the client should update its homeserver URL with the discovery
     /// information present in the login response.
     respect_login_well_known: bool,
@@ -461,6 +467,7 @@ impl ClientInner {
             // A single `RoomUpdates` is sent once per sync, so we assume that 32 is sufficient
             // ballast for all observers to catch up.
             room_updates_sender: broadcast::Sender::new(32),
+            room_updates_sent: AtomicU64::new(0),
             respect_login_well_known,
             sync_beat: event_listener::Event::new(),
             event_cache,
@@ -1281,6 +1288,16 @@ impl Client {
         self.inner.room_updates_sender.subscribe()
     }
 
+    pub(crate) fn room_updates_sent(&self) -> u64 {
+        self.inner.room_updates_sent.load(Ordering::SeqCst)
+    }
+
+    pub(crate) fn send_room_updates(&self, rooms: RoomUpdates) {
+        self.inner.room_updates_sent.fetch_add(1, Ordering::SeqCst);
+        // Ignore errors when there are no receivers.
+        let _ = self.inner.room_updates_sender.send(rooms);
+    }
+
     pub(crate) async fn notification_handlers(
         &self,
     ) -> RwLockReadGuard<'_, Vec<NotificationHandlerFn>> {

Modified file: crates/matrix-sdk/src/event_cache/mod.rs+28 additions−2 deletions@@ -31,6 +31,7 @@ use std::{
     collections::HashMap,
     fmt,
     sync::{Arc, OnceLock, RwLock as StdRwLock, RwLockReadGuard, RwLockWriteGuard},
+    time::Duration,
 };
 
 use futures_util::future::try_join_all;
@@ -42,11 +43,12 @@ use matrix_sdk_base::{
     task_monitor::BackgroundTaskHandle,
     timer,
 };
+use matrix_sdk_common::timeout::timeout;
 use ruma::{OwnedRoomId, RoomId};
 use tokio::sync::{
     Mutex, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock,
     broadcast::{Receiver, Sender, channel},
-    mpsc,
+    mpsc, watch,
 };
 use tracing::{debug, error, instrument, trace};
 
@@ -226,6 +228,7 @@ impl EventCache {
                 config: StdRwLock::new(EventCacheConfig::default()),
                 store: event_cache_store,
                 multiple_room_updates_lock: Default::default(),
+                room_updates_handled: watch::Sender::new(0),
                 by_room: Default::default(),
                 drop_handles: Default::default(),
                 auto_shrink_sender: Default::default(),
@@ -271,9 +274,11 @@ impl EventCache {
             let task_monitor = client.task_monitor();
 
             // Spawn the task that will listen to all the room updates at once.
+            let room_updates = client.subscribe_to_all_room_updates();
+            self.inner.room_updates_handled.send_replace(client.room_updates_sent());
             let listen_updates_task = task_monitor.spawn_infinite_task("event_cache::room_updates_task", tasks::room_updates_task(
                 self.inner.clone(),
-                client.subscribe_to_all_room_updates(),
+                room_updates,
             )).abort_on_drop();
 
             let ignore_user_list_update_task = task_monitor.spawn_infinite_task("event_cache::ignore_user_list_update_task", tasks::ignore_user_list_update_task(
@@ -358,6 +363,23 @@ impl EventCache {
         self.inner.handle_room_updates(updates).await
     }
 
+    /// Wait until every [`RoomUpdates`] the client had sent when called has
+    /// been handled, for at most `limit`.
+    ///
+    /// Returns `false` if `limit` ran out first.
+    pub(crate) async fn wait_for_room_updates(&self, limit: Duration) -> bool {
+        let Ok(client) = self.inner.client() else {
+            return true;
+        };
+        if !self.has_subscribed() {
+            return true;
+        }
+        let target = client.room_updates_sent();
+        let mut handled = self.inner.room_updates_handled.subscribe();
+        let waited = timeout(handled.wait_for(|handled| *handled >= target), limit).await;
+        waited.is_ok()
+    }
+
     /// Check whether [`EventCache::subscribe`] has been called.
     pub fn has_subscribed(&self) -> bool {
         self.inner.drop_handles.get().is_some()
@@ -488,6 +510,10 @@ struct EventCacheInner {
     // TODO: that's the place to add a cross-process lock!
     multiple_room_updates_lock: Mutex<()>,
 
+    /// How many of the client's [`RoomUpdates`] have been handled, counted in
+    /// the client's `room_updates_sent` terms.
+    room_updates_handled: watch::Sender<u64>,
+
     /// Lazily-filled cache of live [`RoomEventCache`], once per room.
     //
     // It's behind an `Arc` to get owned locks.

Modified file: crates/matrix-sdk/src/event_cache/tasks.rs+2 additions@@ -67,6 +67,7 @@ pub(super) async fn room_updates_task(
                         }
                     }
                 }
+                inner.room_updates_handled.send_modify(|handled| *handled += 1);
             }
 
             Err(RecvError::Lagged(num_skipped)) => {
@@ -77,6 +78,7 @@ pub(super) async fn room_updates_task(
                 if let Err(err) = inner.clear_all_rooms().await {
                     error!("when clearing storage after lag in listen_task: {err}");
                 }
+                inner.room_updates_handled.send_modify(|handled| *handled += num_skipped);
             }
 
             Err(RecvError::Closed) => {

Modified file: crates/matrix-sdk/src/sliding_sync/mod.rs+12 additions@@ -53,6 +53,10 @@ pub use self::{builder::*, client::VersionBuilderError, error::*, list::*};
 use self::{cache::restore_sliding_sync_state, client::SlidingSyncResponseProcessor};
 use crate::{Client, Result, config::RequestConfig};
 
+/// How long a rooms response waits for the event cache before its `pos` is
+/// saved anyway.
+const ROOM_UPDATES_WAIT: Duration = Duration::from_secs(10);
+
 /// The Sliding Sync instance.
 ///
 /// It is OK to clone this type as much as you need: cloning it is cheap.
@@ -649,6 +653,14 @@ impl SlidingSync {
                 .handle_response(response, &mut position_guard, requested_required_states)
                 .await?;
 
+            // A `pos` saved before the event cache stores this response's events
+            // resumes past them after a kill, and no gap is left to fetch them.
+            if this.must_process_rooms_response().await
+                && !this.inner.client.event_cache().wait_for_room_updates(ROOM_UPDATES_WAIT).await
+            {
+                warn!("Saving the position before the event cache handled the response");
+            }
+
             this.cache_to_storage(&position_guard).await?;
 
             // Release the position guard lock.

Modified file: crates/matrix-sdk/src/sync.rs+1 additions−2 deletions@@ -188,8 +188,7 @@ impl Client {
         self.handle_sync_events(HandlerKind::Presence, None, presence).await?;
         self.handle_sync_to_device_events(to_device).await?;
 
-        // Ignore errors when there are no receivers.
-        let _ = self.inner.room_updates_sender.send(rooms.clone());
+        self.send_room_updates(rooms.clone());
 
         for (room_id, room_info) in &rooms.joined {
             let Some(room) = self.get_room(room_id) else {

matrix-rust-sdk

Commit f05c136873e9

Commit `f05c136873e9`