matrix-rust-sdk

public

Star0

Fork 0

Watch0

60 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `19e9256c1d96`

tarball

atagen <boss@atagen.co>2026-10-28 06:01 UTC

3 files changed+158-52

sliding-sync: save pos behind the sync loop, without rewinding the live connection

Waiting up to 10s per response for the event cache stalled live updates all through a catch-up, and since the room list shares its pos through the store, every skipped save rewound the running connection to the last saved pos and refetched everything since. A response's pos now waits in HeldPositions until the event cache has handled its room updates, and a background task saves it, so the sync loop never waits. share_pos only adopts a stored pos this instance didn't write itself.

Modified file: crates/matrix-sdk/src/event_cache/mod.rs+11 additions−18 deletions@@ -382,24 +382,17 @@ 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;
-        }
-        if self.inner.dropped_room_updates.load(Ordering::SeqCst) {
-            return false;
-        }
-        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()
+    /// Follows how many of the client's [`RoomUpdates`] have been handled, in
+    /// the client's `room_updates_sent` terms, or `None` if this cache never
+    /// subscribed and so has nothing to catch up on.
+    pub(crate) fn room_updates_handled(&self) -> Option<watch::Receiver<u64>> {
+        self.has_subscribed().then(|| self.inner.room_updates_handled.subscribe())
+    }
+
+    /// Whether a room update was dropped, after which nothing counts as
+    /// handled until a restart.
+    pub(crate) fn dropped_room_updates(&self) -> bool {
+        self.inner.dropped_room_updates.load(Ordering::SeqCst)
     }
 
     /// Check whether [`EventCache::subscribe`] has been called.

Modified file: crates/matrix-sdk/src/sliding_sync/builder.rs+8 additions−4 deletions@@ -11,8 +11,8 @@ use ruma::{OwnedRoomId, api::client::sync::sync_events::v5 as http};
 use tokio::sync::{Mutex as AsyncMutex, RwLock as AsyncRwLock, broadcast::channel};
 
 use super::{
-    Error, ResponseCommitHook, SlidingSync, SlidingSyncInner, SlidingSyncListBuilder,
-    SlidingSyncPositionMarkers, Version, cache::format_storage_key_prefix,
+    Error, HeldPositions, ResponseCommitHook, SlidingSync, SlidingSyncInner,
+    SlidingSyncListBuilder, SlidingSyncPositionMarkers, Version, cache::format_storage_key_prefix,
 };
 use crate::{Client, Result};
 
@@ -295,13 +295,17 @@ impl SlidingSyncBuilder {
 
             lists,
 
-            position: Arc::new(AsyncMutex::new(SlidingSyncPositionMarkers { pos })),
+            position: Arc::new(AsyncMutex::new(SlidingSyncPositionMarkers { pos: pos.clone() })),
+            held_positions: AsyncMutex::new(HeldPositions {
+                pending: Default::default(),
+                stored: pos,
+            }),
+            position_saver_started: AtomicBool::new(false),
 
             room_subscriptions: StdRwLock::new(self.room_subscriptions),
             extensions: self.extensions.unwrap_or_default(),
 
             internal_channel: internal_channel_sender,
-            event_cache_behind: AtomicBool::new(false),
 
             poll_timeout: self.poll_timeout,
             network_timeout: self.network_timeout,

Modified file: crates/matrix-sdk/src/sliding_sync/mod.rs+139 additions−30 deletions@@ -23,7 +23,7 @@ mod error;
 mod list;
 
 use std::{
-    collections::{BTreeMap, btree_map::Entry},
+    collections::{BTreeMap, VecDeque, btree_map::Entry},
     fmt::Debug,
     future::Future,
     sync::{
@@ -48,7 +48,7 @@ use ruma::{
 };
 use tokio::{
     select,
-    sync::{Mutex as AsyncMutex, OwnedMutexGuard, RwLock as AsyncRwLock, broadcast::Sender},
+    sync::{Mutex as AsyncMutex, OwnedMutexGuard, RwLock as AsyncRwLock, broadcast::Sender, watch},
 };
 use tracing::{Instrument, Span, debug, error, info, instrument, trace, warn};
 
@@ -56,14 +56,6 @@ 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 it gives up on
-/// saving its `pos`.
-const ROOM_UPDATES_WAIT: Duration = Duration::from_secs(10);
-
-/// The same wait once the event cache is known to be behind, long enough for a
-/// recovered one to handle the latest response.
-const ROOM_UPDATES_RECHECK: Duration = Duration::from_secs(1);
-
 /// The Sliding Sync instance.
 ///
 /// It is OK to clone this type as much as you need: cloning it is cheap.
@@ -116,6 +108,13 @@ pub(super) struct SlidingSyncInner {
     /// `position` being updated, before sending a new request.
     position: Arc<AsyncMutex<SlidingSyncPositionMarkers>>,
 
+    /// Positions waiting for the event cache before they're saved.
+    held_positions: AsyncMutex<HeldPositions>,
+
+    /// Whether the task saving held positions as the event cache catches up
+    /// has been spawned.
+    position_saver_started: AtomicBool,
+
     /// The lists of this Sliding Sync instance.
     lists: AsyncRwLock<BTreeMap<String, SlidingSyncList>>,
 
@@ -130,10 +129,32 @@ pub(super) struct SlidingSyncInner {
     /// Internal channel used to pass messages between Sliding Sync and other
     /// types.
     internal_channel: Sender<SlidingSyncInternalMessage>,
+}
+
+/// A `pos` saved before the event cache stores its response's events resumes
+/// past them after a kill, with no gap left to fetch them, so each `pos` waits
+/// here until the event cache has handled its room updates.
+#[derive(Debug)]
+pub(super) struct HeldPositions {
+    /// Oldest first, each with the client's `room_updates_sent` count the
+    /// event cache has to reach before it can be saved.
+    pub(super) pending: VecDeque<(u64, SlidingSyncPositionMarkers)>,
 
-    /// Whether the event cache had not handled every response when the last
-    /// one was handled, in which case `pos` was not saved.
-    event_cache_behind: AtomicBool,
+    /// The `pos` this instance last restored or saved.
+    pub(super) stored: Option<String>,
+}
+
+/// How many positions wait at most, as a stalled event cache would otherwise
+/// add one per response. Dropping the oldest only saves a later one later.
+const HELD_POSITIONS: usize = 256;
+
+impl HeldPositions {
+    fn hold(&mut self, target: u64, position: SlidingSyncPositionMarkers) {
+        if self.pending.len() == HELD_POSITIONS {
+            self.pending.pop_front();
+        }
+        self.pending.push_back((target, position));
+    }
 }
 
 impl SlidingSync {
@@ -145,6 +166,46 @@ impl SlidingSync {
         cache::store_sliding_sync_state(self, position).await
     }
 
+    /// Save the newest held position whose room updates are within the
+    /// `handled` count.
+    async fn save_handled_positions(&self, handled: u64) -> Result<()> {
+        let mut held = self.inner.held_positions.lock().await;
+        if self.inner.client.event_cache().dropped_room_updates() {
+            let before = held.pending.len();
+            held.pending.retain(|(target, _)| *target == 0);
+            if held.pending.len() < before {
+                warn!("Not saving the position, the event cache dropped a room update");
+            }
+        }
+        let mut newest = None;
+        while held.pending.front().is_some_and(|(target, _)| *target <= handled) {
+            newest = held.pending.pop_front().map(|(_, position)| position);
+        }
+        if let Some(position) = newest {
+            self.cache_to_storage(&position).await?;
+            held.stored = position.pos;
+        }
+        Ok(())
+    }
+
+    fn start_position_saver(&self, mut handled: watch::Receiver<u64>) {
+        if self.inner.position_saver_started.swap(true, Ordering::SeqCst) {
+            return;
+        }
+        let inner = Arc::downgrade(&self.inner);
+        spawn(async move {
+            while handled.changed().await.is_ok() {
+                let Some(inner) = inner.upgrade() else {
+                    return;
+                };
+                let count = *handled.borrow_and_update();
+                if let Err(error) = (SlidingSync { inner }).save_handled_positions(count).await {
+                    warn!(%error, "Failed to save a held position");
+                }
+            }
+        });
+    }
+
     /// Create a new [`SlidingSyncBuilder`].
     pub fn builder(id: String, client: Client) -> Result<SlidingSyncBuilder, Error> {
         SlidingSyncBuilder::new(id, client)
@@ -473,6 +534,7 @@ impl SlidingSync {
 
         let to_device_enabled = self.inner.extensions.to_device.enabled == Some(true);
 
+        let mut held = self.inner.held_positions.lock().await;
         let restored_fields = if self.inner.share_pos || to_device_enabled {
             restore_sliding_sync_state(&self.inner.client, &self.inner.storage_key).await?
         } else {
@@ -484,21 +546,27 @@ impl SlidingSync {
         let pos = if self.inner.share_pos {
             if let Some(fields) = &restored_fields {
                 // Override the memory one with the database one, for consistency.
-                if fields.pos != position_guard.pos {
+                // Only another process's `pos` though, since the one this instance saved
+                // lags while the event cache catches up, and going back to it refetches
+                // everything since.
+                if fields.pos != held.stored {
                     info!(
                         "Pos from previous request ('{:?}') was different from \
                          pos in database ('{:?}').",
                         position_guard.pos, fields.pos
                     );
                     position_guard.pos = fields.pos.clone();
+                    held.stored = fields.pos.clone();
+                    held.pending.clear();
                 }
-                fields.pos.clone()
+                position_guard.pos.clone()
             } else {
                 position_guard.pos.clone()
             }
         } else {
             position_guard.pos.clone()
         };
+        drop(held);
 
         // When the client sends a request with no `pos`, MSC4186 returns no device
         // lists updates, as it only returns changes since the provided `pos`
@@ -664,26 +732,27 @@ 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. The
-            // event cache can stall for good, so a lagging one leaves the last saved
-            // `pos` in place, and a restart replays from there instead.
-            let stored = !this.must_process_rooms_response().await || {
-                let behind = this.inner.event_cache_behind.load(Ordering::SeqCst);
-                let wait = if behind { ROOM_UPDATES_RECHECK } else { ROOM_UPDATES_WAIT };
-                this.inner.client.event_cache().wait_for_room_updates(wait).await
-            };
-            this.inner.event_cache_behind.store(!stored, Ordering::SeqCst);
-            if stored {
-                this.cache_to_storage(&position_guard).await?;
+            let handled = if this.must_process_rooms_response().await {
+                this.inner.client.event_cache().room_updates_handled()
             } else {
-                warn!("Not saving the position, the event cache hasn't handled every response");
-            }
+                None
+            };
+            let target = handled.as_ref().map_or(0, |_| this.inner.client.room_updates_sent());
+            this.inner.held_positions.lock().await.hold(target, position_guard.clone());
 
             // Release the position guard lock.
             // It means that other responses can be generated and then handled later.
             drop(position_guard);
 
+            match handled {
+                Some(handled) => {
+                    let count = *handled.borrow();
+                    this.start_position_saver(handled);
+                    this.save_handled_positions(count).await?;
+                }
+                None => this.save_handled_positions(u64::MAX).await?,
+            }
+
             debug!("Done handling response");
 
             Ok(updates)
@@ -843,12 +912,16 @@ impl SlidingSync {
 
             // Invalidate in memory.
             position.pos = None;
+            let mut held = self.inner.held_positions.lock().await;
+            held.pending.clear();
 
             // Propagate to disk.
             // Note: this propagates both the sliding sync state and the cached lists'
             // state to disk.
             if let Err(err) = self.cache_to_storage(&position).await {
                 warn!("Failed to invalidate cached sliding sync state: {err}");
+            } else {
+                held.stored = None;
             }
         }
 
@@ -1022,7 +1095,7 @@ mod tests {
 
     use super::{
         SlidingSync, SlidingSyncBuilder, SlidingSyncList, SlidingSyncListBuilder, SlidingSyncMode,
-        cache::restore_sliding_sync_state, http,
+        SlidingSyncPositionMarkers, cache::restore_sliding_sync_state, http,
     };
     use crate::{
         Client, Result,
@@ -1736,6 +1809,42 @@ mod tests {
         Ok(())
     }
 
+    #[cfg(feature = "e2e-encryption")]
+    #[async_test]
+    async fn test_sliding_sync_keeps_its_own_pos_while_the_saved_one_waits() -> Result<()> {
+        let server = MockServer::start().await;
+        let client = logged_in_client(Some(server.uri())).await;
+        let sliding_sync = client.sliding_sync("elephant-sync")?.share_pos().build().await?;
+        let saved = || async {
+            restore_sliding_sync_state(&client, &sliding_sync.inner.storage_key)
+                .await
+                .unwrap()
+                .and_then(|fields| fields.pos)
+        };
+
+        {
+            let mut held = sliding_sync.inner.held_positions.lock().await;
+            held.hold(1, SlidingSyncPositionMarkers { pos: Some("1".to_owned()) });
+            held.hold(2, SlidingSyncPositionMarkers { pos: Some("2".to_owned()) });
+        }
+        sliding_sync.inner.position.lock().await.pos = Some("2".to_owned());
+
+        sliding_sync.save_handled_positions(0).await?;
+        assert_eq!(saved().await, None);
+        let (request, _, _) = sliding_sync.generate_sync_request().await?;
+        assert_eq!(request.pos.as_deref(), Some("2"));
+
+        sliding_sync.save_handled_positions(1).await?;
+        assert_eq!(saved().await.as_deref(), Some("1"));
+        let (request, _, _) = sliding_sync.generate_sync_request().await?;
+        assert_eq!(request.pos.as_deref(), Some("2"));
+
+        sliding_sync.save_handled_positions(5).await?;
+        assert_eq!(saved().await.as_deref(), Some("2"));
+
+        Ok(())
+    }
+
     #[async_test]
     async fn test_stop_sync_loop() -> Result<()> {
         let (_server, sliding_sync) = new_sliding_sync(vec![

matrix-rust-sdk

Commit 19e9256c1d96

Commit `19e9256c1d96`