matrix-rust-sdk

public

Star0

Fork 0

Watch0

25 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `17b53943be2d`

tarball

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

2 files changed+43-27

timeline: fence history flushes on the subscriber task instead of replacing

flush_history ran wait_for_history and then replace_with_initial_history, so every flush rebuilt the timeline and dropped read receipts and item ids although commit_history_candidate already broadcasts incremental diffs. The subscriber task now drains what is queued after wait_for_history through handle_history_update before replying, which orders the fence with the updates it covers, and only a Lagged receiver still replaces.

Modified file: crates/matrix-sdk-ui/src/timeline/mod.rs+5 additions−1 deletions@@ -227,7 +227,11 @@ pub struct AttachmentConfig {
 }
 
 impl Timeline {
-    #[doc = "Wait until this live timeline has applied all history currently retained by its room cache."]
+    /// Wait for earlier history operations on the room cache, then until this
+    /// live timeline has applied every update they published.
+    ///
+    /// Updates apply incrementally, keeping read receipts and item ids, and
+    /// only a lagged subscriber falls back to replacing the timeline.
     pub async fn flush_history(&self) -> Result<(), Error> {
         let Some(flush) = &self.history_flush else {
             return Ok(());

Modified file: crates/matrix-sdk-ui/src/timeline/tasks.rs+38 additions−26 deletions@@ -25,7 +25,10 @@ use matrix_sdk::{
     send_queue::RoomSendQueueUpdate,
 };
 use ruma::OwnedEventId;
-use tokio::sync::broadcast::{Receiver, error::RecvError};
+use tokio::sync::broadcast::{
+    Receiver,
+    error::{RecvError, TryRecvError},
+};
 use tokio::sync::{mpsc, oneshot};
 use tracing::{error, instrument, trace, warn};
 
@@ -213,39 +216,48 @@ pub(in crate::timeline) async fn room_event_cache_updates_task(
 ) {
     trace!("Spawned the event subscriber task.");
 
+    let mut flushing: Option<oneshot::Sender<Result<(), crate::timeline::Error>>> = None;
+
     loop {
         trace!("Waiting for an event.");
 
-        let update = tokio::select! {
-            request = async {
-                match history_flush_requests.as_mut() {
-                    Some(requests) => requests.recv().await,
-                    None => std::future::pending::<Option<oneshot::Sender<Result<(), crate::timeline::Error>>>>().await,
+        let update = if let Some(reply) = flushing.take() {
+            match room_event_cache_subscriber.try_recv() {
+                Ok(update) => {
+                    flushing = Some(reply);
+                    Ok(update)
+                }
+                Err(TryRecvError::Empty) => {
+                    let _ = reply.send(Ok(()));
+                    continue;
                 }
-            } => {
-                if let Some(reply) = request {
-                    if let Err(error) = room_event_cache.wait_for_history().await {
-                        let _ = reply.send(Err(error.into()));
-                        continue;
+                Err(TryRecvError::Lagged(num_skipped)) => {
+                    flushing = Some(reply);
+                    Err(RecvError::Lagged(num_skipped))
+                }
+                Err(TryRecvError::Closed) => break,
+            }
+        } else {
+            tokio::select! {
+                request = async {
+                    match history_flush_requests.as_mut() {
+                        Some(requests) => requests.recv().await,
+                        None => std::future::pending().await,
                     }
-                    match room_event_cache.subscribe_with_history().await {
-                        Ok((initial_events, gaps, subscriber)) => {
-                            room_event_cache_subscriber = subscriber;
-                            timeline_controller
-                                .replace_with_initial_history(initial_events, gaps, RemoteEventOrigin::Cache)
-                                .await;
-                            let _ = reply.send(Ok(()));
-                        }
-                        Err(error) => {
-                            let _ = reply.send(Err(error.into()));
-                        }
+                } => {
+                    match request {
+                        Some(reply) => match room_event_cache.wait_for_history().await {
+                            Ok(()) => flushing = Some(reply),
+                            Err(error) => {
+                                let _ = reply.send(Err(error.into()));
+                            }
+                        },
+                        None => history_flush_requests = None,
                     }
-                } else {
-                    history_flush_requests = None;
+                    continue;
                 }
-                continue;
+                update = room_event_cache_subscriber.recv() => update,
             }
-            update = room_event_cache_subscriber.recv() => update,
         };
         let update = match update {
             Ok(up) => up,

matrix-rust-sdk

Commit 17b53943be2d

Commit `17b53943be2d`