From 30353d3b4e677ea57e598dede4f5d32948aa22f6 Mon Sep 17 00:00:00 2001 From: Ivan Enderlin Date: Fri, 24 Apr 2026 15:16:47 +0200 Subject: [PATCH] refactor(sdk): Move threads from `RoomEventCache` to `Caches`. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This patch starts flattening caches by starting the extract thread caches from the room cache. This patch comments some methods or adds a `todo!()` to make the code compile. It's the price for smaller commits. The thread cache gains new methods like `handle_joined_room_update` or `handle_joined_left_update`. It mimics the flow in the room cache. A new `caches::aggregator` module is created to introduce “pipelines”: a `Timeline` is created for each cache. It lives in front of the room cache, and basically reverses the “post process” flow to a “pre process” flow. More commits will improve it. `ResetCaches` is now also responsible to clear the threads in addition to the room (previously, the room cache was responsible of that). --- .../src/event_cache/caches/aggregator.rs | 75 +++++++ .../matrix-sdk/src/event_cache/caches/mod.rs | 196 ++++++++++++++++-- .../src/event_cache/caches/room/mod.rs | 49 ++--- .../src/event_cache/caches/room/state.rs | 14 +- .../src/event_cache/caches/room/updates.rs | 2 +- .../src/event_cache/caches/thread/mod.rs | 69 +++++- .../src/event_cache/caches/thread/state.rs | 17 +- 7 files changed, 355 insertions(+), 67 deletions(-) create mode 100644 crates/matrix-sdk/src/event_cache/caches/aggregator.rs diff --git a/crates/matrix-sdk/src/event_cache/caches/aggregator.rs b/crates/matrix-sdk/src/event_cache/caches/aggregator.rs new file mode 100644 index 000000000..1266c151d --- /dev/null +++ b/crates/matrix-sdk/src/event_cache/caches/aggregator.rs @@ -0,0 +1,75 @@ +// Copyright 2026 The Matrix.org Foundation C.I.C. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashMap; + +use matrix_sdk_base::{ + serde_helpers::{extract_edit_target, extract_thread_root}, + sync::Timeline, +}; +use ruma::OwnedEventId; + +use super::{super::Result, room::RoomEventCacheStateLockReadGuard, thread::ThreadEventCache}; + +pub fn aggregate_timeline_for_room(timeline: Timeline) -> Timeline { + timeline +} + +pub async fn aggregate_timeline_for_threads( + timeline: &Timeline, + existing_threads: &HashMap, + room_event_cache: RoomEventCacheStateLockReadGuard<'_>, +) -> Result> { + let mut new_events_by_thread = HashMap::new(); + + let default_timeline = || Timeline { + limited: timeline.limited, + prev_batch: timeline.prev_batch.clone(), + events: Vec::new(), + }; + + for event in &timeline.events { + // This event is part of a thread. + if let Some(thread_root) = extract_thread_root(event.raw()) { + new_events_by_thread + .entry(thread_root) + .or_insert_with(default_timeline) + .events + .push(event.clone()); + } + // This event is the root of a thread. + else if let Some(event_id) = event.event_id() + && existing_threads.contains_key(&event_id) + { + new_events_by_thread + .entry(event_id) + .or_insert_with(default_timeline) + .events + .push(event.clone()); + } + + // This event is an edit that may apply to a thread.. + if let Some(edit_target) = extract_edit_target(event.raw()) { + // This event is known and part of a thread. + if let Some((_location, edit_target_event)) = + room_event_cache.find_event(&edit_target).await? + && let Some(thread_root) = extract_thread_root(edit_target_event.raw()) + { + new_events_by_thread.entry(thread_root).or_insert_with(default_timeline); + } + } + } + + Ok(new_events_by_thread) +} diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index 4a29e842a..318fdf966 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::{collections::HashMap, ops::Deref, sync::Arc}; + use eyeball::SharedObservable; use eyeball_im::VectorDiff; use matrix_sdk_base::{ @@ -20,12 +22,13 @@ use matrix_sdk_base::{ linked_chunk::Position, sync::{JoinedRoomUpdate, LeftRoomUpdate}, }; -use ruma::{OwnedRoomId, RoomId}; -use tokio::sync::{broadcast::Sender, mpsc}; +use ruma::{OwnedEventId, OwnedRoomId, RoomId}; +use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock, broadcast::Sender, mpsc}; use super::{EventCacheError, EventsOrigin, Result, automatic_pagination::AutomaticPagination}; use crate::{client::WeakClient, room::WeakRoom}; +mod aggregator; pub mod event_focused; pub mod event_linked_chunk; pub(super) mod lock; @@ -39,6 +42,14 @@ pub mod thread; #[derive(Debug)] pub(super) struct Caches { pub room: room::RoomEventCache, + pub threads: Arc>>, + internals: CachesInternals, +} + +#[derive(Debug)] +struct CachesInternals { + store: EventCacheStoreLock, + linked_chunk_update_sender: Sender, } impl Caches { @@ -76,14 +87,14 @@ impl Caches { client.user_id().expect("the user must be logged in, at this point").to_owned(); let room_state = room::LockedRoomEventCacheState::new( - own_user_id, + own_user_id.clone(), room_id.to_owned(), weak_room.clone(), room_version_rules, enabled_thread_support, update_sender.clone(), - linked_chunk_update_sender, - store, + linked_chunk_update_sender.clone(), + store.clone(), pagination_status.clone(), automatic_pagination, ) @@ -95,6 +106,7 @@ impl Caches { let room_event_cache = room::RoomEventCache::new( room_id.to_owned(), weak_room, + own_user_id, room_state, pagination_status, auto_shrink_sender, @@ -108,23 +120,135 @@ impl Caches { .send(room::RoomEventCacheGenericUpdate { room_id: room_id.to_owned() }); } - Ok(Self { room: room_event_cache }) + Ok(Self { + room: room_event_cache, + threads: Arc::new(RwLock::new(HashMap::new())), + internals: CachesInternals { store, linked_chunk_update_sender }, + }) + } + + /// Get the [`RoomEventCache`]. + /// + /// [`RoomEventCache`]: room::RoomEventCache + pub async fn room(&self) -> &room::RoomEventCache { + &self.room + } + + /// Get or create a [`ThreadEventCache`]. + /// + /// Note: it is impossible to know if `thread_id` represents a valid thread + /// identifier. It means it's possible to create a [`ThreadEventCache`] for + /// an event that is not a thread root. + /// + /// [`ThreadEventCache`]: thread::ThreadEventCache + pub async fn thread( + &self, + thread_id: OwnedEventId, + ) -> Result< + OwnedRwLockReadGuard< + HashMap, + thread::ThreadEventCache, + >, + > { + Ok( + match OwnedRwLockWriteGuard::try_downgrade_map( + self.threads.clone().write_owned().await, + |threads| threads.get(&thread_id), + ) { + // Thread exists. + Ok(locked_cache) => locked_cache, + // Thread does not exist, let's create it. + Err(mut threads) => { + let room = &self.room; + let cache = thread::ThreadEventCache::new( + room.room_id().to_owned(), + thread_id.clone(), + room.own_user_id().to_owned(), + room.weak_room().to_owned(), + self.internals.store.clone(), + room.update_sender().generic_update_sender().clone(), + self.internals.linked_chunk_update_sender.clone(), + ) + .await?; + + threads.insert(thread_id.clone(), cache); + + OwnedRwLockWriteGuard::downgrade_map(threads, |threads| { + threads.get(&thread_id).unwrap() + }) + } + }, + ) } /// Update all the event caches with a [`JoinedRoomUpdate`]. pub(super) async fn handle_joined_room_update(&self, updates: JoinedRoomUpdate) -> Result<()> { - let Self { room } = &self; + let Self { room, threads, internals: _ } = &self; - room.handle_joined_room_update(updates).await?; + // Room. + { + let mut updates = updates.clone(); + updates.timeline = aggregator::aggregate_timeline_for_room(updates.timeline); + + room.handle_joined_room_update(updates).await?; + } + + // Threads. + { + let mut updates = updates.clone(); + updates.account_data.clear(); + updates.ambiguity_changes.clear(); + + let timeline_for_threads = aggregator::aggregate_timeline_for_threads( + &updates.timeline, + threads.read().await.deref(), + room.state().read().await?, + ) + .await?; + + for (thread_id, timeline) in timeline_for_threads { + let mut updates = updates.clone(); + updates.timeline = timeline; + + self.thread(thread_id).await?.handle_joined_room_update(updates).await?; + } + } Ok(()) } /// Update all the event caches with a [`LeftRoomUpdate`]. pub(super) async fn handle_left_room_update(&self, updates: LeftRoomUpdate) -> Result<()> { - let Self { room } = &self; + let Self { room, threads, internals: _ } = &self; - room.handle_left_room_update(updates).await?; + // Room. + { + let mut updates = updates.clone(); + updates.timeline = aggregator::aggregate_timeline_for_room(updates.timeline); + + room.handle_left_room_update(updates).await?; + } + + // Threads. + { + let mut updates = updates.clone(); + updates.account_data.clear(); + updates.ambiguity_changes.clear(); + + let timeline_for_threads = aggregator::aggregate_timeline_for_threads( + &updates.timeline, + threads.read().await.deref(), + room.state().read().await?, + ) + .await?; + + for (thread_id, timeline) in timeline_for_threads { + let mut updates = updates.clone(); + updates.timeline = timeline; + + self.thread(thread_id).await?.handle_left_room_update(updates).await?; + } + } Ok(()) } @@ -157,15 +281,33 @@ impl Caches { /// To reset all the event caches, call [`ResetCaches::reset_all`]. If this type /// is dropped, no reset happens and the exclusive lock is released. pub(super) struct ResetCaches<'c> { - room_lock: (&'c room::RoomEventCache, room::RoomEventCacheStateLockWriteGuard<'c>), + room_lock: (room::RoomEventCacheStateLockWriteGuard<'c>, room::RoomEventCacheUpdateSender), + threads_lock: OwnedRwLockWriteGuard>, + thread_locks: Vec<( + thread::OwnedThreadEventCacheStateLockWriteGuard, + thread::ThreadEventCacheUpdateSender, + )>, } impl<'c> ResetCaches<'c> { /// Create a new [`ResetCaches`]. /// /// It can fail if acquiring an exclusive lock fails. - async fn new(Caches { room }: &'c mut Caches) -> Result { - Ok(Self { room_lock: (room, room.state().write().await?) }) + async fn new(Caches { room, threads, internals: _ }: &'c mut Caches) -> Result { + // Acquire an exclusive access to the state of the room. + let room_lock = (room.state().write().await?, room.update_sender().clone()); + + // Acquire an exclusive access to the threads. + // Then, for each thread, acquire an exclusive access to its state. + let threads_lock = threads.clone().write_owned().await; + let mut thread_locks = Vec::new(); + + for thread in threads_lock.values() { + thread_locks + .push((thread.state().write_owned().await?, thread.update_sender().clone())); + } + + Ok(Self { room_lock, threads_lock, thread_locks }) } /// Reset all the event caches, and broadcast the [`TimelineVectorDiffs`]. @@ -175,19 +317,41 @@ impl<'c> ResetCaches<'c> { /// /// It can fail if resetting an event cache fails. pub async fn reset_all(self) -> Result<()> { - let Self { room_lock: (room, mut room_state) } = self; + let Self { room_lock, threads_lock, thread_locks } = self; { + let (mut room_state, room_update_sender) = room_lock; + let updates_as_vector_diffs = room_state.reset().await?; - room.update_sender().send( + room_update_sender.send( room::RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs: updates_as_vector_diffs, origin: EventsOrigin::Cache, }), - Some(room::RoomEventCacheGenericUpdate { room_id: room.room_id().to_owned() }), + Some(room::RoomEventCacheGenericUpdate { room_id: room_state.room_id.clone() }), ); } + { + for thread_lock in thread_locks { + let (mut thread_state, thread_update_sender) = thread_lock; + + let updates_as_vector_diffs = thread_state.reset().await?; + thread_update_sender.send( + TimelineVectorDiffs { + diffs: updates_as_vector_diffs, + origin: EventsOrigin::Cache, + }, + // This function is part of the `RoomEventCache` flow. The generic update is + // handled by it. + None, + ); + } + + // Now we can release the exclusive acces over the threads. + drop(threads_lock); + } + Ok(()) } } diff --git a/crates/matrix-sdk/src/event_cache/caches/room/mod.rs b/crates/matrix-sdk/src/event_cache/caches/room/mod.rs index d9619b4c2..c944afd48 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/mod.rs @@ -37,7 +37,9 @@ use ruma::{ use tokio::sync::{Notify, broadcast::Receiver, mpsc}; use tracing::{instrument, trace, warn}; -pub(super) use self::state::{LockedRoomEventCacheState, RoomEventCacheStateLockWriteGuard}; +pub(super) use self::state::{ + LockedRoomEventCacheState, RoomEventCacheStateLockReadGuard, RoomEventCacheStateLockWriteGuard, +}; pub use self::{ subscriber::RoomEventCacheSubscriber, updates::{ @@ -79,20 +81,23 @@ impl RoomEventCache { pub(super) fn new( room_id: OwnedRoomId, weak_room: WeakRoom, + own_user_id: OwnedUserId, state: LockedRoomEventCacheState, shared_pagination_status: SharedObservable, auto_shrink_sender: mpsc::Sender, update_sender: RoomEventCacheUpdateSender, ) -> Self { Self { - inner: Arc::new(RoomEventCacheInner::new( + inner: Arc::new(RoomEventCacheInner { room_id, weak_room, + own_user_id, state, - shared_pagination_status, - auto_shrink_sender, update_sender, - )), + pagination_batch_token_notifier: Notify::new(), + auto_shrink_sender, + shared_pagination_status, + }), } } @@ -101,6 +106,16 @@ impl RoomEventCache { &self.inner.room_id } + /// Get the owner of this [`RoomEventCache`]. + pub(super) fn own_user_id(&self) -> &OwnedUserId { + &self.inner.own_user_id + } + + /// Get the weak room of this [`RoomEventCache`]. + pub(super) fn weak_room(&self) -> &WeakRoom { + &self.inner.weak_room + } + /// Read all current events. /// /// Use [`RoomEventCache::subscribe`] to get all current events, plus a @@ -443,6 +458,9 @@ pub(super) struct RoomEventCacheInner { weak_room: WeakRoom, + /// The user's own user id. + own_user_id: OwnedUserId, + /// State for this room's event cache. state: LockedRoomEventCacheState, @@ -462,27 +480,6 @@ pub(super) struct RoomEventCacheInner { } impl RoomEventCacheInner { - /// Creates a new cache for a room, and subscribes to room updates, so as - /// to handle new timeline events. - fn new( - room_id: OwnedRoomId, - weak_room: WeakRoom, - state: LockedRoomEventCacheState, - shared_pagination_status: SharedObservable, - auto_shrink_sender: mpsc::Sender, - update_sender: RoomEventCacheUpdateSender, - ) -> Self { - Self { - room_id, - weak_room, - state, - update_sender, - pagination_batch_token_notifier: Notify::new(), - auto_shrink_sender, - shared_pagination_status, - } - } - fn handle_account_data(&self, account_data: Vec>) { if account_data.is_empty() { return; diff --git a/crates/matrix-sdk/src/event_cache/caches/room/state.rs b/crates/matrix-sdk/src/event_cache/caches/room/state.rs index b580f4cd5..4feea32fd 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -13,8 +13,7 @@ // limitations under the License. use std::{ - collections::{BTreeMap, HashMap, HashSet, hash_map::Entry}, - ops::DerefMut, + collections::{BTreeMap, HashMap, HashSet}, sync::{ Arc, OnceLock, atomic::{AtomicUsize, Ordering}, @@ -785,15 +784,6 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { async fn reset_internal(&mut self) -> Result<(), EventCacheError> { self.state.room_linked_chunk.reset(); - - // No need to update the thread summaries: the room events are - // gone because of the reset of `room_linked_chunk`. - // - // Clear the threads. - for thread in self.state.threads.values_mut() { - thread.clear().await?; - } - self.propagate_changes().await?; // Reset the pagination state too: pretend we never waited for the initial @@ -1059,6 +1049,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { Ok(()) } + /* pub(super) async fn get_or_reload_thread( &mut self, root_event_id: OwnedEventId, @@ -1093,6 +1084,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { Entry::Occupied(entry) => Ok(entry.into_mut()), } } + */ #[instrument(skip_all)] async fn update_threads( diff --git a/crates/matrix-sdk/src/event_cache/caches/room/updates.rs b/crates/matrix-sdk/src/event_cache/caches/room/updates.rs index e7b40b4dd..4814d4649 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/updates.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/updates.rs @@ -139,7 +139,7 @@ impl RoomEventCacheUpdateSender { } /// Get the generic update sender. - pub(super) fn generic_update_sender(&self) -> &Sender { + pub(in super::super) fn generic_update_sender(&self) -> &Sender { &self.generic_sender } diff --git a/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs b/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs index 4f8531103..0453a0bd2 100644 --- a/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs @@ -20,16 +20,22 @@ mod updates; use std::{fmt, sync::Arc}; -use matrix_sdk_base::event_cache::{Event, store::EventCacheStoreLock}; +use matrix_sdk_base::{ + event_cache::{Event, store::EventCacheStoreLock}, + sync::{JoinedRoomUpdate, LeftRoomUpdate, Timeline}, +}; use ruma::{EventId, OwnedEventId, OwnedRoomId, OwnedUserId}; use tokio::sync::{ Notify, broadcast::{Receiver, Sender}, }; -use tracing::{error, trace}; +use tracing::{error, instrument, trace}; -pub(super) use self::state::LockedThreadEventCacheState; -use self::{pagination::ThreadPagination, updates::ThreadEventCacheUpdateSender}; +use self::pagination::ThreadPagination; +pub(super) use self::{ + state::{LockedThreadEventCacheState, OwnedThreadEventCacheStateLockWriteGuard}, + updates::ThreadEventCacheUpdateSender, +}; use super::{ super::Result, EventsOrigin, TimelineVectorDiffs, @@ -39,7 +45,7 @@ use super::{ use crate::room::WeakRoom; /// All the information related to a single thread. -pub(super) struct ThreadEventCache { +pub struct ThreadEventCache { inner: Arc, } @@ -135,13 +141,53 @@ impl ThreadEventCache { ThreadPagination::new(self.inner.clone()) } - /// Clear a thread. - pub async fn clear(&mut self) -> Result<()> { - let updates_as_vector_diffs = self.inner.state.write().await?.reset().await?; + /// Return a reference to the state. + pub(in super::super) fn state(&self) -> &LockedThreadEventCacheState { + &self.inner.state + } - if !updates_as_vector_diffs.is_empty() { + /// Get a reference to the [`RoomEventCacheUpdateSender`]. + pub(in super::super) fn update_sender(&self) -> &ThreadEventCacheUpdateSender { + &self.inner.update_sender + } + + /// Handle a [`JoinedRoomUpdate`]. + #[instrument(skip_all, fields(room_id = %self.inner.room_id, thread_root = %self.inner.thread_id))] + pub(super) async fn handle_joined_room_update(&self, updates: JoinedRoomUpdate) -> Result<()> { + self.handle_timeline(updates.timeline).await?; + + Ok(()) + } + + /// Handle a [`LeftRoomUpdate`]. + #[instrument(skip_all, fields(room_id = %self.inner.room_id, thread_root = %self.inner.thread_id))] + pub(super) async fn handle_left_room_update(&self, updates: LeftRoomUpdate) -> Result<()> { + self.handle_timeline(updates.timeline).await?; + + Ok(()) + } + + /// Handle a [`Timeline`], i.e. new events received by a sync for this + /// thread. + async fn handle_timeline(&self, timeline: Timeline) -> Result<()> { + if timeline.events.is_empty() && timeline.prev_batch.is_none() { + return Ok(()); + } + + trace!("adding new events"); + + let (stored_prev_batch_token, timeline_event_diffs) = + self.inner.state.write().await?.handle_sync(timeline).await?; + + // Now that all events have been added, we can trigger the + // `pagination_token_notifier`. + if stored_prev_batch_token { + self.inner.pagination_batch_token_notifier.notify_one(); + } + + if !timeline_event_diffs.is_empty() { self.inner.update_sender.send( - TimelineVectorDiffs { diffs: updates_as_vector_diffs, origin: EventsOrigin::Cache }, + TimelineVectorDiffs { diffs: timeline_event_diffs, origin: EventsOrigin::Sync }, // This function is part of the `RoomEventCache` flow. The generic update is // handled by it. None, @@ -158,6 +204,8 @@ impl ThreadEventCache { events: Vec, prev_batch_token: &Option, ) -> Result<()> { + todo!() + /* if events.is_empty() { return Ok(()); } @@ -183,6 +231,7 @@ impl ThreadEventCache { } Ok(()) + */ } /// Replaces a single event, be it saved in memory or in the store. diff --git a/crates/matrix-sdk/src/event_cache/caches/thread/state.rs b/crates/matrix-sdk/src/event_cache/caches/thread/state.rs index c693eb9b7..9986a8412 100644 --- a/crates/matrix-sdk/src/event_cache/caches/thread/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/thread/state.rs @@ -21,6 +21,7 @@ use matrix_sdk_base::{ linked_chunk::{ ChunkIdentifierGenerator, LinkedChunkId, OwnedLinkedChunkId, Position, Update, lazy_loader, }, + sync::Timeline, }; use matrix_sdk_common::executor::spawn; use ruma::{EventId, OwnedEventId, OwnedRoomId, OwnedUserId}; @@ -343,9 +344,10 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> { #[must_use = "Propagate `VectorDiff` updates via `TimelineVectorDiffs`"] pub async fn handle_sync( &mut self, - events: Vec, - prev_batch_token: &Option, + timeline: Timeline, ) -> Result<(bool, Vec>)> { + let prev_batch_token = &timeline.prev_batch; + let DeduplicationOutcome { all_events: events, in_memory_duplicated_event_ids, @@ -356,7 +358,7 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> { &self.store, LinkedChunkId::Thread(&self.state.room_id, &self.state.thread_id), &self.state.thread_linked_chunk, - events, + timeline.events, ) .await?; @@ -387,6 +389,15 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> { self.state.propagate_changes(&self.store).await?; + if timeline.limited && has_new_gap { + // If there was a previous batch token for a limited timeline, unload the chunks + // so it only contains the last one; otherwise, there might be a + // valid gap in between, and observers may not render it (yet). + // + // We must do this *after* persisting these events to storage. + self.state.shrink_to_last_chunk(&self.store).await?; + } + let timeline_event_diffs = self.state.thread_linked_chunk.updates_as_vector_diffs(); Ok((has_new_gap, timeline_event_diffs))