From a4633ae467c709449a4e21ce809659dc23094bbc Mon Sep 17 00:00:00 2001 From: Stefan Ceriu Date: Sun, 26 Jul 2026 11:22:07 +0300 Subject: [PATCH] feat(event cache): wire the queue into the event cache, replacing the credit system --- bindings/matrix-sdk-ffi/src/client.rs | 2 +- crates/matrix-sdk/changelog.d/6805.changed.md | 1 + crates/matrix-sdk/changelog.d/6805.feature.md | 1 + .../src/event_cache/automatic_pagination.rs | 347 ------------------ .../matrix-sdk/src/event_cache/caches/mod.rs | 6 +- .../src/event_cache/caches/read_receipts.rs | 24 +- .../src/event_cache/caches/room/state.rs | 12 +- crates/matrix-sdk/src/event_cache/mod.rs | 77 ++-- .../integration/event_cache/read_receipts.rs | 2 +- labs/multiverse/src/main.rs | 2 +- 10 files changed, 62 insertions(+), 412 deletions(-) create mode 100644 crates/matrix-sdk/changelog.d/6805.changed.md create mode 100644 crates/matrix-sdk/changelog.d/6805.feature.md delete mode 100644 crates/matrix-sdk/src/event_cache/automatic_pagination.rs diff --git a/bindings/matrix-sdk-ffi/src/client.rs b/bindings/matrix-sdk-ffi/src/client.rs index da664d1fe..29c2ce846 100644 --- a/bindings/matrix-sdk-ffi/src/client.rs +++ b/bindings/matrix-sdk-ffi/src/client.rs @@ -2316,7 +2316,7 @@ impl Client { /// This must be called after creating a client, but before subscribing to /// the event cache (so, before spawning a sync service or a timeline). pub fn enable_automatic_backpagination(&self) { - self.inner.event_cache().config_mut().experimental_auto_backpagination = true; + self.inner.event_cache().config_mut().experimental_auto_back_pagination = true; } pub fn homeserver_capabilities(&self) -> HomeserverCapabilities { diff --git a/crates/matrix-sdk/changelog.d/6805.changed.md b/crates/matrix-sdk/changelog.d/6805.changed.md new file mode 100644 index 000000000..43b07f3a3 --- /dev/null +++ b/crates/matrix-sdk/changelog.d/6805.changed.md @@ -0,0 +1 @@ +[**breaking**] Background back-pagination is now driven by the shared `BackPaginationQueue` instead of a per-room credit system. `EventCacheConfig::room_pagination_per_room_credit` (and `DEFAULT_ROOM_PAGINATION_CREDITS`) are removed, replaced by `EventCacheConfig::max_concurrent_back_paginations`. As part of this, the latest event of a room is now computed with automatic back-pagination: when no suitable candidate is in memory, the room's history is back-paginated until a candidate is found or the start of the timeline is reached. Requires `EventCacheConfig::experimental_auto_back_pagination`. diff --git a/crates/matrix-sdk/changelog.d/6805.feature.md b/crates/matrix-sdk/changelog.d/6805.feature.md new file mode 100644 index 000000000..89d3809c0 --- /dev/null +++ b/crates/matrix-sdk/changelog.d/6805.feature.md @@ -0,0 +1 @@ +The event cache gained a shared `BackPaginationQueue` (`EventCache::back_pagination_queue`): a single background executor that runs back-pagination requests from every use case (search backfill, latest-event resolution, read-receipt finding) by `Priority`, with a bounded in-flight number (`EventCacheConfig::max_concurrent_back_paginations`) to protect the server, and one run per room at a time. `BackPaginationQueue::run_search_backfill(BackPaginationStrategy)` sweeps every room down to a ~3-month floor (front-loaded by recency) to populate the search index. diff --git a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs deleted file mode 100644 index d908b802d..000000000 --- a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs +++ /dev/null @@ -1,347 +0,0 @@ -// 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, - sync::{Arc, Weak}, -}; - -use matrix_sdk_base::task_monitor::{BackgroundTaskHandle, TaskMonitor}; -use ruma::{OwnedRoomId, RoomId}; -use tokio::sync::mpsc; -use tracing::{info, instrument, trace, warn}; - -use crate::event_cache::EventCacheInner; - -/// State for running paginations in background tasks. -/// -/// Shallow type, can be cloned cheaply. -#[derive(Clone)] -pub struct AutomaticPagination { - inner: Arc, -} - -#[cfg(not(tarpaulin_include))] -impl std::fmt::Debug for AutomaticPagination { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("AutomaticPagination").finish_non_exhaustive() - } -} - -impl AutomaticPagination { - /// Create a new [`AutomaticPagination`], spawning the background task to - /// handle incoming requests to run background paginations. - pub(super) fn new(event_cache: Weak, task_monitor: &TaskMonitor) -> Self { - let (sender, receiver) = mpsc::unbounded_channel(); - - let task = task_monitor.spawn_infinite_task( - "event_cache::automatic_paginations_task", - automatic_paginations_task(event_cache, receiver), - ); - - Self { inner: Arc::new(AutomaticPaginationInner { _task: task, sender }) } - } - - /// Request a single back-pagination to happen in the background for the - /// given room. - /// - /// Returns false, if the request couldn't be sent. - pub fn run_once(&self, room_id: &RoomId) -> bool { - // We don't want to do anything with the error type, as it only includes the - // request we just created, and not much more; there's no guarantee that - // retrying sending it would succeed, so let it drop, and report the - // result as a boolean, for informative purposes. - self.inner - .sender - .send(AutomaticPaginationRequest::PaginateRoomBackwards { room_id: room_id.to_owned() }) - .is_ok() - } -} - -struct AutomaticPaginationInner { - /// The task used to handle automatic pagination requests. - _task: BackgroundTaskHandle, - - /// A sender for automatic pagination requests, that is shared with every - /// room. - /// - /// It's a `OnceLock` because its initialization is deferred to - /// [`EventCache::subscribe`]. - sender: mpsc::UnboundedSender, -} - -#[derive(Clone, Debug)] -enum AutomaticPaginationRequest { - PaginateRoomBackwards { room_id: OwnedRoomId }, -} - -/// Listen to background automatic pagination requests, and execute them in -/// real-time. -#[instrument(skip_all)] -async fn automatic_paginations_task( - inner: Weak, - mut receiver: mpsc::UnboundedReceiver, -) { - trace!("Spawning the automatic pagination task"); - - let mut room_pagination_credits = HashMap::new(); - - while let Some(request) = receiver.recv().await { - match request { - AutomaticPaginationRequest::PaginateRoomBackwards { room_id } => { - let Some(inner) = inner.upgrade() else { - // The event cache has been dropped, exit the task. - break; - }; - - let config = *inner.config.read().unwrap(); - - let credits = room_pagination_credits - .entry(room_id.clone()) - .or_insert(config.room_pagination_per_room_credit); - - if *credits == 0 { - trace!(for_room = %room_id, "No more credits to paginate this room in the background, skipping"); - continue; - } - - let pagination = match inner.all_caches_for_room(&room_id).await { - Ok(caches) => caches.room.pagination(), - Err(err) => { - warn!(for_room = %room_id, "Failed to get the `Caches`: {err}"); - continue; - } - }; - - trace!(for_room = %room_id, "automatic backpagination triggered"); - - match pagination.run_backwards_once(config.room_pagination_batch_size).await { - Ok(outcome) => { - // Pagination requests must be idempotent, so we only decrement credits if - // we actually paginated something new. - if !outcome.reached_start || !outcome.events.is_empty() { - *credits -= 1; - } - } - - Err(err) => { - warn!(for_room = %room_id, "Failed to run background pagination: {err}"); - // Don't decrement credits in this case, to allow a - // retry later. - } - } - } - } - } - - // The sender has shut down, exit. - info!("Closing the automatic pagination task because receiver closed"); -} - -// MatrixMockServer et al. aren't available on wasm. -#[cfg(all(test, not(target_arch = "wasm32")))] -mod tests { - use std::time::Duration; - - use assert_matches::assert_matches; - use eyeball_im::VectorDiff; - use matrix_sdk_base::sleep::sleep; - use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; - use ruma::{event_id, room_id}; - - use crate::{ - assert_let_timeout, - event_cache::{EventsOrigin, RoomEventCacheUpdate}, - test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, - }; - - /// Test that we can send automatic pagination requests. - #[async_test] - async fn test_background_room_paginations() { - let server = MatrixMockServer::new().await; - let client = server.client_builder().build().await; - - let event_cache = client.event_cache(); - event_cache.config_mut().experimental_auto_backpagination = true; - event_cache.subscribe().unwrap(); - - let room_id = room_id!("!omelette:fromage.fr"); - let f = EventFactory::new().room(room_id).sender(*BOB); - - let room = server.sync_joined_room(&client, room_id).await; - - let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap(); - - // Starting with an empty, inactive room, - let (room_events, mut room_cache_updates) = room_event_cache.subscribe().await.unwrap(); - assert!(room_events.is_empty()); - assert!(room_cache_updates.is_empty()); - - // We get a gappy sync (so as to have a previous-batch token), - server - .sync_room( - &client, - JoinedRoomBuilder::new(room_id) - .set_timeline_limited() - .set_timeline_prev_batch("prev_batch"), - ) - .await; - - { - assert_let_timeout!( - Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() - ); - assert_eq!(update.diffs.len(), 1); - assert_matches!(update.diffs[0], VectorDiff::Clear); - assert_matches!(update.origin, EventsOrigin::Sync); - } - - // Set up the mock for /messages, - server - .mock_room_messages() - .ok(RoomMessagesResponseTemplate::default().events(vec![ - f.text_msg("comté").event_id(event_id!("$2")), - f.text_msg("beaufort").event_id(event_id!("$1")), - ])) - .mock_once() - .mount() - .await; - - // Send a request for a background pagination, - let automatic_pagination_api = event_cache.automatic_pagination().unwrap(); - assert!(automatic_pagination_api.run_once(room_id)); - - // The room pagination happens in the background. - assert_let_timeout!( - Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() - ); - assert_eq!(update.diffs.len(), 1); - - assert_matches!(update.origin, EventsOrigin::Pagination); - - let mut room_events = room_events.into(); - for diff in update.diffs { - diff.apply(&mut room_events); - } - - assert_eq!(room_events.len(), 2); - assert_eq!(room_events[0].event_id().unwrap(), event_id!("$1")); - assert_eq!(room_events[1].event_id().unwrap(), event_id!("$2")); - - // And there's no more updates. - assert!(room_cache_updates.is_empty()); - } - - /// Test that the credit system works. - #[async_test] - async fn test_room_pagination_respects_credits_system() { - let server = MatrixMockServer::new().await; - let client = server.client_builder().build().await; - - let event_cache = client.event_cache(); - event_cache.config_mut().experimental_auto_backpagination = true; - - // Only allow 1 background pagination per room, to test that the credit system - // is properly taken into account. - event_cache.config_mut().room_pagination_per_room_credit = 1; - event_cache.subscribe().unwrap(); - - let room_id = room_id!("!omelette:fromage.fr"); - let f = EventFactory::new().room(room_id).sender(*BOB); - - let room = server.sync_joined_room(&client, room_id).await; - let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap(); - - // Starting with an empty, inactive room, - let (room_events, mut room_cache_updates) = room_event_cache.subscribe().await.unwrap(); - assert!(room_events.is_empty()); - assert!(room_cache_updates.is_empty()); - - server - .sync_room( - &client, - JoinedRoomBuilder::new(room_id) - .set_timeline_limited() - .set_timeline_prev_batch("prev_batch"), - ) - .await; - - { - assert_let_timeout!( - Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() - ); - assert_eq!(update.diffs.len(), 1); - assert_matches!(update.diffs[0], VectorDiff::Clear); - assert_matches!(update.origin, EventsOrigin::Sync); - } - - // Set up the mock for /messages, so that it returns another prev-batch token, - server - .mock_room_messages() - .match_from("prev_batch") - .ok(RoomMessagesResponseTemplate::default() - .events(vec![ - f.text_msg("comté").event_id(event_id!("$2")), - f.text_msg("beaufort").event_id(event_id!("$1")), - ]) - .end_token("prev_batch_2")) - .mock_once() - .mount() - .await; - - // Send a request for a background pagination, - let automatic_pagination_api = event_cache.automatic_pagination().unwrap(); - assert!(automatic_pagination_api.run_once(room_id)); - - // The room pagination happens in the background. - assert_let_timeout!( - Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() - ); - assert_eq!(update.diffs.len(), 1); - - assert_matches!(update.origin, EventsOrigin::Pagination); - - let mut room_events = room_events.into(); - for diff in update.diffs { - diff.apply(&mut room_events); - } - - assert_eq!(room_events.len(), 2); - assert_eq!(room_events[0].event_id().unwrap(), event_id!("$1")); - assert_eq!(room_events[1].event_id().unwrap(), event_id!("$2")); - - // And there's no more updates yet. - assert!(room_cache_updates.is_empty()); - - // One can send another request to back-paginate… - assert!(automatic_pagination_api.run_once(room_id)); - - sleep(Duration::from_millis(300)).await; - // But it doesn't happen, because we don't have enough credits for automatic - // backpagination. - assert!(room_cache_updates.is_empty()); - - // We can still manually backpaginate with success, though. - server - .mock_room_messages() - .match_from("prev_batch_2") - .ok(RoomMessagesResponseTemplate::default()) - .mock_once() - .mount() - .await; - - let outcome = room_event_cache.pagination().run_backwards_once(30).await.unwrap(); - assert!(outcome.reached_start); - } -} diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index 7875412da..f0ab78bf3 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -29,7 +29,7 @@ use tokio::sync::{ use self::subscriber::AutoShrinkMessage; use super::{ - EventCacheError, EventsOrigin, Result, automatic_pagination::AutomaticPagination, states, + EventCacheError, EventsOrigin, Result, back_pagination_queue::BackPaginationQueue, states, }; use crate::{client::WeakClient, room::WeakRoom}; @@ -90,7 +90,7 @@ impl Caches { linked_chunk_update_sender: Sender, auto_shrink_sender: mpsc::Sender, state: &states::StateLock, - automatic_pagination: Option, + back_pagination_queue: Option, ) -> Result { let Some(client) = weak_client.get() else { return Err(EventCacheError::ClientDropped); @@ -129,7 +129,7 @@ impl Caches { linked_chunk_update_sender.clone(), store_guard, pagination_status.clone(), - automatic_pagination, + back_pagination_queue, ) }, ) diff --git a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs index b458aa81a..8294c319f 100644 --- a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs +++ b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs @@ -99,6 +99,8 @@ //! [`RoomEventCache`]: super::room::RoomEventCache //! [`ThreadEventCache`]: super::thread::ThreadEventCache +use std::collections::HashSet; + use matrix_sdk_base::{ read_receipts::{LatestReadReceipt, RoomReadReceipts}, serde_helpers::extract_relation, @@ -120,7 +122,8 @@ use ruma::{ use tracing::{debug, instrument, trace, warn}; use super::{ - super::automatic_pagination::AutomaticPagination, event_linked_chunk::EventLinkedChunk, + super::back_pagination_queue::{BackPaginationQueue, stop_on_event_ids}, + event_linked_chunk::EventLinkedChunk, }; trait RoomReadReceiptsExt { @@ -390,7 +393,7 @@ pub(crate) async fn compute_unread_counts( linked_chunk: &EventLinkedChunk, read_receipts: &mut RoomReadReceipts, with_threading_support: bool, - automatic_pagination: Option<&AutomaticPagination>, + back_pagination_queue: Option<&BackPaginationQueue>, state_store: &DynStateStore, ) { debug!(?read_receipts, "Starting"); @@ -434,13 +437,16 @@ pub(crate) async fn compute_unread_counts( } // Request a pagination: we haven't found a better receipt, but we haven't even - // found the latest active receipt! - if let Some(automatic_pagination) = automatic_pagination { - if automatic_pagination.run_once(room_id) { - trace!("Requested pagination to find a better receipt"); - } else { - warn!("Failed to request pagination to find a better receipt"); - } + // found the latest active receipt! Hand it the receipt event ids we're chasing + // so the backfill can stop as soon as one of them is loaded. + if let Some(back_pagination_queue) = back_pagination_queue { + let targets: HashSet = read_receipts + .pending + .iter() + .cloned() + .chain(read_receipts.latest_active.as_ref().map(|receipt| receipt.event_id.clone())) + .collect(); + back_pagination_queue.paginate_for_read_receipt(room_id, stop_on_event_ids(targets)); } // If we haven't returned at this point, it means we don't have any new "active" 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 793001523..b4360fb47 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -43,7 +43,7 @@ use super::{ super::{ super::{ EventCacheError, - automatic_pagination::AutomaticPagination, + back_pagination_queue::BackPaginationQueue, deduplicator::{DeduplicationOutcome, filter_duplicate_events}, persistence::{ find_event, find_event_relations, find_event_with_relations, @@ -102,8 +102,8 @@ pub struct RoomEventCacheState { /// A handle for subscribers. subscribers_handle: SubscribersHandle, - /// A copy of the automatic pagination API object. - automatic_pagination: Option, + /// A handle to the shared back-pagination queue. + back_pagination_queue: Option, } impl RoomEventCacheState { @@ -128,7 +128,7 @@ impl RoomEventCacheState { linked_chunk_update_sender: Sender, store_guard: EventCacheStoreLockGuard, pagination_status: SharedObservable, - automatic_pagination: Option, + back_pagination_queue: Option, ) -> Result { let linked_chunk_id = LinkedChunkId::Room(&room_id); @@ -188,7 +188,7 @@ impl RoomEventCacheState { room_version_rules, waited_for_initial_prev_token: false, subscribers_handle: Default::default(), - automatic_pagination, + back_pagination_queue, }) } @@ -656,7 +656,7 @@ impl<'a> StateLockWriteGuard<'a, RoomEventCacheState> { &self.state.room_linked_chunk, &mut read_receipts, self.state.enabled_thread_support, - self.state.automatic_pagination.as_ref(), + self.state.back_pagination_queue.as_ref(), room.client().state_store(), ) .await; diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index 3e9274aff..a676fd8e1 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -55,7 +55,6 @@ use crate::{ paginators::PaginatorError, }; -mod automatic_pagination; mod back_pagination_queue; mod caches; mod deduplicator; @@ -68,8 +67,9 @@ mod tasks; #[cfg(feature = "e2e-encryption")] pub use redecryptor::{DecryptionRetryRequest, RedecryptorReport}; +pub(crate) use self::back_pagination_queue::RoomBackPaginationEnd; pub use self::{ - automatic_pagination::AutomaticPagination, + back_pagination_queue::{BackPaginationQueue, BackPaginationStrategy}, caches::{ TimelineVectorDiffs, event_focused::{EventFocusThreadMode, EventFocusedCache, EventFocusedCacheKey}, @@ -265,7 +265,7 @@ impl EventCache { linked_chunk_update_sender, #[cfg(feature = "e2e-encryption")] redecryption_channels, - automatic_pagination: OnceLock::new(), + back_pagination_queue: OnceLock::new(), thread_subscriber_sender, }), } @@ -360,13 +360,19 @@ impl EventCache { ) .abort_on_drop(); - if self.config().experimental_auto_backpagination { - // Run the deferred initialization of the automatic pagination request sender, that - // is shared with every room. - trace!("spawning the automatic paginations API"); - self.inner.automatic_pagination.get_or_init(|| AutomaticPagination::new(Arc::downgrade(&self.inner), task_monitor)); + if self.config().experimental_auto_back_pagination { + // Deferred initialization of the shared back-pagination queue. + trace!("spawning the back-pagination queue"); + let max_concurrent = self.config().max_concurrent_back_paginations; + self.inner.back_pagination_queue.get_or_init(|| { + BackPaginationQueue::new( + Arc::downgrade(&self.inner), + max_concurrent, + task_monitor, + ) + }); } else { - trace!("automatic paginations API is disabled"); + trace!("back-pagination queue is disabled"); } Arc::new(EventCacheDropHandles { @@ -489,11 +495,10 @@ impl EventCache { self.inner.generic_update_sender.subscribe() } - /// Returns a reference to the [`AutomaticPagination`] API, if enabled at - /// construction with the - /// [`EventCacheConfig::experimental_auto_backpagination`] flag. - pub fn automatic_pagination(&self) -> Option { - self.inner.automatic_pagination.get().cloned() + /// Returns the shared [`BackPaginationQueue`], if enabled at construction + /// with the [`EventCacheConfig::experimental_auto_back_pagination`] flag. + pub fn back_pagination_queue(&self) -> Option { + self.inner.back_pagination_queue.get().cloned() } } @@ -509,23 +514,14 @@ pub struct EventCacheConfig { /// Whether to automatically backpaginate a room under certain conditions. /// /// Off by default. - pub experimental_auto_backpagination: bool, + pub experimental_auto_back_pagination: bool, - /// The maximum number of allowed room paginations, for a given room, that - /// can be executed in the automatic paginations task. + /// The maximum number of back-paginations the background queue runs at + /// once, across all rooms and use cases. Bounds server load. /// - /// After that number of paginations, the task will stop executing - /// paginations for that room *in the background* (user-requested - /// paginations will still be executed, of course). - /// - /// Defaults to [`EventCacheConfig::DEFAULT_ROOM_PAGINATION_CREDITS`]. - pub room_pagination_per_room_credit: usize, - - /// The number of messages to paginate in a single batch, when executing an - /// automatic pagination request. - /// - /// Defaults to [`EventCacheConfig::DEFAULT_ROOM_PAGINATION_BATCH_SIZE`]. - pub room_pagination_batch_size: u16, + /// Defaults to + /// [`EventCacheConfig::DEFAULT_MAX_CONCURRENT_BACK_PAGINATIONS`]. + pub max_concurrent_back_paginations: usize, } impl EventCacheConfig { @@ -536,15 +532,9 @@ impl EventCacheConfig { /// loading the pinned events. pub const DEFAULT_MAX_CONCURRENT_REQUESTS: usize = 8; - /// The default number of credits to give to a room for automatic - /// paginations (see also - /// [`EventCacheConfig::room_pagination_per_room_credit`]). - pub const DEFAULT_ROOM_PAGINATION_CREDITS: usize = 20; - - /// The default number of messages to paginate in a single batch, when - /// executing an automatic pagination request (see also - /// [`EventCacheConfig::room_pagination_batch_size`]). - pub const DEFAULT_ROOM_PAGINATION_BATCH_SIZE: u16 = 30; + /// The default maximum number of concurrent background back-paginations + /// (see also [`EventCacheConfig::max_concurrent_back_paginations`]). + pub const DEFAULT_MAX_CONCURRENT_BACK_PAGINATIONS: usize = 3; } impl Default for EventCacheConfig { @@ -552,9 +542,8 @@ impl Default for EventCacheConfig { Self { max_pinned_events_concurrent_requests: Self::DEFAULT_MAX_CONCURRENT_REQUESTS, max_pinned_events_to_load: Self::DEFAULT_MAX_EVENTS_TO_LOAD, - room_pagination_per_room_credit: Self::DEFAULT_ROOM_PAGINATION_CREDITS, - room_pagination_batch_size: Self::DEFAULT_ROOM_PAGINATION_BATCH_SIZE, - experimental_auto_backpagination: false, + max_concurrent_back_paginations: Self::DEFAULT_MAX_CONCURRENT_BACK_PAGINATIONS, + experimental_auto_back_pagination: false, } } } @@ -619,9 +608,9 @@ struct EventCacheInner { /// State for the automatic pagination mechanism. /// - /// Depends on the [`EventCacheConfig::experimental_auto_backpagination`] + /// Depends on the [`EventCacheConfig::experimental_auto_back_pagination`] /// flag to be set at subscription time. - automatic_pagination: OnceLock, + back_pagination_queue: OnceLock, } impl EventCacheInner { @@ -760,7 +749,7 @@ impl EventCacheInner { "we must have called `EventCache::subscribe()` before calling here.", ), &self.state, - self.automatic_pagination.get().cloned(), + self.back_pagination_queue.get().cloned(), ) .await?; diff --git a/crates/matrix-sdk/tests/integration/event_cache/read_receipts.rs b/crates/matrix-sdk/tests/integration/event_cache/read_receipts.rs index 7d63523a9..0a685f6fe 100644 --- a/crates/matrix-sdk/tests/integration/event_cache/read_receipts.rs +++ b/crates/matrix-sdk/tests/integration/event_cache/read_receipts.rs @@ -727,7 +727,7 @@ async fn test_compute_unread_counts_triggers_backpaginations() { let client = server.client_builder().build().await; let own_user_id = client.user_id().unwrap(); - client.event_cache().config_mut().experimental_auto_backpagination = true; + client.event_cache().config_mut().experimental_auto_back_pagination = true; client.event_cache().subscribe().unwrap(); let room_id = room_id!("!omelette:fromage.fr"); diff --git a/labs/multiverse/src/main.rs b/labs/multiverse/src/main.rs index 67a1f5ef4..9e69783c3 100644 --- a/labs/multiverse/src/main.rs +++ b/labs/multiverse/src/main.rs @@ -150,7 +150,7 @@ async fn main() -> Result<()> { }); let event_cache = client.event_cache(); - event_cache.config_mut().experimental_auto_backpagination = true; + event_cache.config_mut().experimental_auto_back_pagination = true; event_cache.subscribe()?; let terminal = ratatui::init();