From 45bdf830ca8f77255c51c8ea6ab0cbb09588dc79 Mon Sep 17 00:00:00 2001 From: Ivan Enderlin Date: Fri, 20 Mar 2026 15:16:47 +0100 Subject: [PATCH] feat(sdk): Create `ThreadEventCacheUpdateSender`. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This patch creates the `ThreadEventCacheUpdateSender` à la `RoomEventCacheUpdateSender` to abstract the sender. This patch uses `ThreadEventCacheUpdateSender` in `ThreadEventCacheInner` (new) along with `ThreadEventCacheState`. The use in the `ThreadEventCacheInner` type is necessary to be able to send updates without reaching the state (which is behind an async faillible lock). This is necessary for pagination, see next commit. --- .../src/event_cache/caches/thread/mod.rs | 22 ++++++---- .../event_cache/caches/thread/pagination.rs | 6 +-- .../src/event_cache/caches/thread/state.rs | 35 ++++++++++------ .../src/event_cache/caches/thread/updates.rs | 40 +++++++++++++++++++ 4 files changed, 79 insertions(+), 24 deletions(-) create mode 100644 crates/matrix-sdk/src/event_cache/caches/thread/updates.rs 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 3356f3f35..ee85a57db 100644 --- a/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs @@ -16,6 +16,7 @@ pub mod pagination; mod state; +mod updates; use std::{fmt, sync::Arc}; @@ -25,7 +26,7 @@ pub(super) use state::LockedThreadEventCacheState; use tokio::sync::broadcast::{Receiver, Sender}; use tracing::{error, trace}; -use self::pagination::ThreadPagination; +use self::{pagination::ThreadPagination, updates::ThreadEventCacheUpdateSender}; use super::{ super::Result, EventsOrigin, TimelineVectorDiffs, room::RoomEventCacheLinkedChunkUpdate, }; @@ -46,6 +47,9 @@ struct ThreadEventCacheInner { /// State for this thread's event cache. state: LockedThreadEventCacheState, + + /// Update sender for this room. + update_sender: ThreadEventCacheUpdateSender, } impl fmt::Debug for ThreadEventCache { @@ -64,6 +68,8 @@ impl ThreadEventCache { store: EventCacheStoreLock, linked_chunk_update_sender: Sender, ) -> Result { + let update_sender = ThreadEventCacheUpdateSender::new(); + Ok(Self { inner: Arc::new(ThreadEventCacheInner { thread_id: thread_id.clone(), @@ -73,9 +79,11 @@ impl ThreadEventCache { thread_id, own_user_id, store, + update_sender.clone(), linked_chunk_update_sender, ) .await?, + update_sender, }), }) } @@ -87,7 +95,7 @@ impl ThreadEventCache { let events = state.thread_linked_chunk().events().map(|(_position, item)| item.clone()).collect(); - let recv = state.state.sender.subscribe(); + let recv = self.inner.update_sender.new_thread_receiver(); Ok((events, recv)) } @@ -100,12 +108,10 @@ impl ThreadEventCache { /// Clear a thread, after a gappy sync for instance. pub async fn clear(&mut self) -> Result<()> { - let mut state = self.inner.state.write().await?; - - let updates_as_vector_diffs = state.reset().await?; + let updates_as_vector_diffs = self.inner.state.write().await?.reset().await?; if !updates_as_vector_diffs.is_empty() { - let _ = state.state.sender.send(TimelineVectorDiffs { + let _ = self.inner.update_sender.send(TimelineVectorDiffs { diffs: updates_as_vector_diffs, origin: EventsOrigin::Cache, }); @@ -127,7 +133,7 @@ impl ThreadEventCache { let timeline_event_diffs = state.handle_sync(events).await?; if !timeline_event_diffs.is_empty() { - let _ = state.state.sender.send(TimelineVectorDiffs { + let _ = self.inner.update_sender.send(TimelineVectorDiffs { diffs: timeline_event_diffs, origin: EventsOrigin::Sync, }); @@ -158,7 +164,7 @@ impl ThreadEventCache { let timeline_event_diffs = state.thread_linked_chunk_mut().updates_as_vector_diffs(); if !timeline_event_diffs.is_empty() { - let _ = state.state.sender.send(TimelineVectorDiffs { + let _ = self.inner.update_sender.send(TimelineVectorDiffs { diffs: timeline_event_diffs, origin: EventsOrigin::Sync, }); diff --git a/crates/matrix-sdk/src/event_cache/caches/thread/pagination.rs b/crates/matrix-sdk/src/event_cache/caches/thread/pagination.rs index ab47cdc14..c64965816 100644 --- a/crates/matrix-sdk/src/event_cache/caches/thread/pagination.rs +++ b/crates/matrix-sdk/src/event_cache/caches/thread/pagination.rs @@ -276,9 +276,9 @@ impl PaginatedCache for ThreadEventCacheWrapper { if !updates.is_empty() { // Send the updates to the listeners. - let _ = state - .state - .sender + let _ = self + .cache + .update_sender .send(TimelineVectorDiffs { diffs: updates, origin: EventsOrigin::Pagination }); } 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 a8956fd7e..fd54c56a6 100644 --- a/crates/matrix-sdk/src/event_cache/caches/thread/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/thread/state.rs @@ -27,16 +27,19 @@ use ruma::{OwnedEventId, OwnedRoomId, OwnedUserId}; use tokio::sync::broadcast::Sender; use tracing::{debug, error, instrument}; -use super::super::{ +use super::{ super::{ - EventCacheError, EventsOrigin, Result, - deduplicator::{DeduplicationOutcome, filter_duplicate_events}, - persistence::{load_linked_chunk_metadata, send_updates_to_store}, + super::{ + EventCacheError, EventsOrigin, Result, + deduplicator::{DeduplicationOutcome, filter_duplicate_events}, + persistence::{load_linked_chunk_metadata, send_updates_to_store}, + }, + TimelineVectorDiffs, + event_linked_chunk::{EventLinkedChunk, sort_positions_descending}, + lock, + room::RoomEventCacheLinkedChunkUpdate, }, - TimelineVectorDiffs, - event_linked_chunk::{EventLinkedChunk, sort_positions_descending}, - lock, - room::RoomEventCacheLinkedChunkUpdate, + ThreadEventCacheUpdateSender, }; pub struct ThreadEventCacheState { @@ -56,8 +59,11 @@ pub struct ThreadEventCacheState { /// The linked chunk for this thread. thread_linked_chunk: EventLinkedChunk, - /// A sender for live events updates in this thread. - pub sender: Sender, + /// A clone of [`super::ThreadEventCacheInner::update_sender`]. + /// + /// This is used only by the [`ThreadEventCacheStateLock::read`] and + /// [`ThreadEventCacheStateLock::write`] when the state must be reset. + update_sender: ThreadEventCacheUpdateSender, /// A sender for the globally observable linked chunk updates that happened /// during a sync or a back-pagination. @@ -100,6 +106,7 @@ impl LockedThreadEventCacheState { thread_id: OwnedEventId, own_user_id: OwnedUserId, store: EventCacheStoreLock, + update_sender: ThreadEventCacheUpdateSender, linked_chunk_update_sender: Sender, ) -> Result { let store_guard = match store.lock().await? { @@ -167,7 +174,7 @@ impl LockedThreadEventCacheState { linked_chunk, full_linked_chunk_metadata, ), - sender: Sender::new(32), + update_sender, linked_chunk_update_sender, waited_for_initial_prev_token: false, })) @@ -194,8 +201,10 @@ impl<'a> lock::Reload for ThreadEventCacheStateLockWriteGuard<'a> { let diffs = self.state.thread_linked_chunk.updates_as_vector_diffs(); if !diffs.is_empty() { - let _ = - self.state.sender.send(TimelineVectorDiffs { diffs, origin: EventsOrigin::Cache }); + let _ = self + .state + .update_sender + .send(TimelineVectorDiffs { diffs, origin: EventsOrigin::Cache }); } Ok(()) diff --git a/crates/matrix-sdk/src/event_cache/caches/thread/updates.rs b/crates/matrix-sdk/src/event_cache/caches/thread/updates.rs new file mode 100644 index 000000000..f042ba51a --- /dev/null +++ b/crates/matrix-sdk/src/event_cache/caches/thread/updates.rs @@ -0,0 +1,40 @@ +// 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 tokio::sync::broadcast::{Receiver, Sender}; + +use crate::event_cache::TimelineVectorDiffs; + +/// A small type to send updates in all channels. +#[derive(Clone)] +pub struct ThreadEventCacheUpdateSender { + thread_sender: Sender, +} + +impl ThreadEventCacheUpdateSender { + /// Create a new [`ThreadEventCacheUpdateSender`]. + pub fn new() -> Self { + Self { thread_sender: Sender::new(32) } + } + + /// Send a [`TimelineVectorDiffs`]. + pub fn send(&self, thread_update: TimelineVectorDiffs) { + let _ = self.thread_sender.send(thread_update); + } + + /// Create a new [`Receiver`] of [`TimelineVectorDiffs`]. + pub(super) fn new_thread_receiver(&self) -> Receiver { + self.thread_sender.subscribe() + } +}