Merge pull request #6344 from matrix-org/bnjbvr/automatic-backpagination-unread-counts

feat(event cache): automatic back-pagination for unread counts, MVP
2026-04-01 09:26:55 +02:00
parent 4071a55cd7 06e6dd05c3
commit 388ced09a6
11 changed files with 613 additions and 53 deletions
@@ -2155,6 +2155,15 @@ impl Client {
        }))))
    }

+    /// Whether to enable automatic backpagination under certain conditions
+    /// (e.g. when processing read receipts).
+    ///
+    /// This is an experimental feature, and might cause performance issues on
+    /// large accounts. Use with caution.
+    pub fn enable_automatic_backpagination(&self) {
+        self.inner.event_cache().config_mut().experimental_auto_backpagination = true;
+    }
+
    pub fn homeserver_capabilities(&self) -> HomeserverCapabilities {
        HomeserverCapabilities::new(self.inner.homeserver_capabilities())
    }
@@ -389,7 +389,7 @@ async fn test_max_events_to_load_is_honored() {
    let client = server.client_builder().build().await;
    let room_id = room_id!("!test:localhost");

-    client.event_cache().config_mut().await.max_pinned_events_to_load = 1;
+    client.event_cache().config_mut().max_pinned_events_to_load = 1;

    let f = EventFactory::new().room(room_id).sender(*BOB);
    let pinned_event = f
@@ -869,7 +869,7 @@ async fn test_ensure_max_concurrency_is_observed() {
    let max_concurrent_requests = 10;

    // Define the max concurrent requests allowed for the event cache.
-    client.event_cache().config_mut().await.max_pinned_events_concurrent_requests =
+    client.event_cache().config_mut().max_pinned_events_concurrent_requests =
        max_concurrent_requests;

    let f = EventFactory::new().room(&room_id).sender(user_id!("@example:localhost"));
@@ -0,0 +1,347 @@
+// 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<AutomaticPaginationInner>,
+}
+
+#[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<EventCacheInner>, task_monitor: &TaskMonitor) -> Self {
+        let (sender, receiver) = mpsc::unbounded_channel();
+
+        let task = task_monitor.spawn_background_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<AutomaticPaginationRequest>,
+}
+
+#[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<EventCacheInner>,
+    mut receiver: mpsc::UnboundedReceiver<AutomaticPaginationRequest>,
+) {
+    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);
+    }
+}
@@ -24,7 +24,9 @@ use ruma::{OwnedRoomId, RoomId};
 use tokio::sync::{broadcast::Sender, mpsc};

 use super::{EventCacheError, EventsOrigin, Result};
-use crate::{client::WeakClient, room::WeakRoom};
+use crate::{
+    client::WeakClient, event_cache::automatic_pagination::AutomaticPagination, room::WeakRoom,
+};

 pub mod event_focused;
 pub mod event_linked_chunk;
@@ -50,6 +52,7 @@ impl Caches {
        linked_chunk_update_sender: Sender<room::RoomEventCacheLinkedChunkUpdate>,
        auto_shrink_sender: mpsc::Sender<OwnedRoomId>,
        store: EventCacheStoreLock,
+        automatic_pagination: Option<AutomaticPagination>,
    ) -> Result<Self> {
        let Some(client) = weak_client.get() else {
            return Err(EventCacheError::ClientDropped);
@@ -84,6 +87,7 @@ impl Caches {
            linked_chunk_update_sender,
            store,
            pagination_status.clone(),
+            automatic_pagination,
        )
        .await?;

@@ -463,7 +463,7 @@ impl PinnedEventCache {
    async fn reload_pinned_events(room: Room) -> Result<Option<Vec<Event>>> {
        let (max_events_to_load, max_concurrent_requests) = {
            let client = room.client();
-            let config = client.event_cache().config().await;
+            let config = client.event_cache().config();
            (config.max_pinned_events_to_load, config.max_pinned_events_concurrent_requests)
        };

@@ -110,7 +110,9 @@ use ruma::{
 };
 use tracing::{debug, instrument, trace, warn};

-use crate::event_cache::caches::event_linked_chunk::EventLinkedChunk;
+use crate::event_cache::{
+    automatic_pagination::AutomaticPagination, caches::event_linked_chunk::EventLinkedChunk,
+};

 trait RoomReadReceiptsExt {
    /// Update the [`RoomReadReceipts`] unread counts according to the new
@@ -219,6 +221,20 @@ impl RoomReadReceiptsExt for RoomReadReceipts {
    }
 }

+struct SelectBestReceiptResult {
+    /// Whether to request a back-pagination after trying to select the best
+    /// receipt.
+    ///
+    /// This is usually a signal that we didn't find any better receipt, or the
+    /// latest active receipt, in the current in-memory linked chunk, and
+    /// that we need to fetch more historical data (from disk or from the
+    /// network) to get a better picture of the unread counts.
+    request_pagination: bool,
+
+    /// The event id of the best receipt we found, if any.
+    receipt: Option<OwnedEventId>,
+}
+
 /// Return a new better (i.e. more recent) receipt based on a search of the
 /// linked chunk.
 ///
@@ -235,7 +251,7 @@ fn select_best_receipt(
    new_receipt_event: Option<&ReceiptEventContent>,
    latest_active: Option<&EventId>,
    with_threading_support: bool,
-) -> Option<OwnedEventId> {
+) -> SelectBestReceiptResult {
    // If we had a new receipt event, add the main/unthreaded receipts it contains
    // to the pending receipts list. We'll try to chase them later.
    if let Some(receipt_event) = new_receipt_event {
@@ -263,17 +279,17 @@ fn select_best_receipt(
    // i.e., we've found a better receipt, *and* there's no more pending receipt
    // to try to match against events in the linked chunk.

-    let mut found = None;
+    let mut result = SelectBestReceiptResult { request_pagination: false, receipt: None };

    for (event, event_id) in
        linked_chunk.revents().filter_map(|(_pos, ev)| Some((ev, ev.event_id()?)))
    {
-        if found.is_none() {
+        if result.receipt.is_none() {
            // Try to see if the latest active receipt is still the most recent receipt.
            if latest_active == Some(&event_id) {
                // The latest active receipt is still the most recent receipt, so keep it.
                trace!(active = %event_id, "the latest active receipt is still the most recent; stopping search");
-                found = Some(event_id.clone());
+                result.receipt = Some(event_id.clone());
            }
            // Try to find an implicit read receipt (i.e. an event sent by the current
            // user).
@@ -284,14 +300,14 @@ fn select_best_receipt(
                && (!with_threading_support || extract_thread_root(event.raw()).is_none())
            {
                trace!(implicit = %event_id, "found an implicit receipt; stopping search");
-                found = Some(event_id.clone());
+                result.receipt = Some(event_id.clone());
            }
        }

        // Early exit condition (see the comment above): we've already found a most
        // recent receipt, and there's no other pending receipts to match against known
        // events.
-        if found.is_some() && pending_receipts.is_empty() {
+        if result.receipt.is_some() && pending_receipts.is_empty() {
            trace!("exiting loop; found a better receipt, and no more pending receipt to match");
            break;
        }
@@ -301,9 +317,9 @@ fn select_best_receipt(
        // one!
        pending_receipts.retain(|pending| {
            if *pending == event_id {
-                if found.is_none() {
+                if result.receipt.is_none() {
                    trace!(pending = %event_id, "found a pending receipt; stopping search");
-                    found = Some(event_id.clone());
+                    result.receipt = Some(event_id.clone());
                } else {
                    trace!(%event_id, "discarding a pending receipt that wasn't selected");
                }
@@ -318,13 +334,22 @@ fn select_best_receipt(
        });
    }

-    found.or_else(|| latest_active.map(|event_id| {
-        // The only time when we can run into this is because we didn't find a more recent receipt,
-        // and the previously active one wasn't loaded in the linked chunk. In this case, every
-        // event currently loaded in the linked chunk will be processed in `process_event()`.
-        trace!(%event_id, "reusing previous active receipt (but we didn't find it in the linked chunk)");
-        event_id.to_owned()
-    }))
+    if result.receipt.is_none() {
+        // Request a patination: we haven't found a better receipt, but we haven't even
+        // found the latest active receipt!
+        result.request_pagination = true;
+
+        // In the meanwhile, reuse the latest active receipt, if any.
+        result.receipt = latest_active.map(|event_id| {
+            // The only time when we can run into this is because we didn't find a more recent receipt,
+            // and the previously active one wasn't loaded in the linked chunk. In this case, every
+            // event currently loaded in the linked chunk will be processed in `process_event()`.
+            trace!(%event_id, "reusing previous active receipt (but we didn't find it in the linked chunk)");
+            event_id.to_owned()
+        });
+    }
+
+    result
 }

 /// Given a set of events coming from sync, for a room, update the
@@ -340,30 +365,37 @@ pub(crate) fn compute_unread_counts(
    linked_chunk: &EventLinkedChunk,
    read_receipts: &mut RoomReadReceipts,
    with_threading_support: bool,
+    automatic_pagination: Option<&AutomaticPagination>,
 ) {
    debug!(?read_receipts, "Starting");

-    let new_receipt = select_best_receipt(
+    let select_best_receipt_result = select_best_receipt(
        user_id,
        linked_chunk,
        &mut read_receipts.pending,
        receipt_event,
        read_receipts.latest_active.as_ref().map(|latest_active| latest_active.event_id.as_ref()),
        with_threading_support,
-    )
-    .map(|event_id| LatestReadReceipt { event_id });
+    );

-    if let Some(new_receipt) = new_receipt {
+    if select_best_receipt_result.request_pagination {
+        trace!("Requesting pagination to find a better receipt");
+        if let Some(automatic_pagination) = automatic_pagination
+            && !automatic_pagination.run_once(room_id)
+        {
+            warn!("Failed to request pagination to find a better receipt");
+        }
+    }
+
+    if let Some(event_id) = select_best_receipt_result.receipt {
        // We've found the id of an event to which the receipt attaches. The associated
        // event may either come from the new batch of events associated to
        // this sync, or it may live in the past timeline events we know
        // about.

-        let event_id = new_receipt.event_id.clone();
-
        // First, save the event id as the latest one that has a read receipt.
        trace!(%event_id, "Saving a new active read receipt");
-        read_receipts.latest_active = Some(new_receipt);
+        read_receipts.latest_active = Some(LatestReadReceipt { event_id: event_id.clone() });

        // The event for the receipt is in the linked chunk, so we'll find it and can
        // count safely from here.
@@ -844,9 +876,11 @@ mod tests {
            active_receipt,
            with_threading_support,
        );
-        assert!(result.is_none());
+        assert!(result.receipt.is_none());
        // And there are no pending receipts.
        assert!(pending_receipts.is_empty());
+        // In this case, select_best_receipt will request a pagination.
+        assert!(result.request_pagination);
    }

    #[test]
@@ -881,9 +915,11 @@ mod tests {
            active_receipt,
            with_threading_support,
        );
-        assert_eq!(result.unwrap(), event_id!("$2"));
+        assert_eq!(result.receipt.unwrap(), event_id!("$2"));
        // And there are no pending receipts.
        assert!(pending_receipts.is_empty());
+        // We don't need to paginate, as we found a receipt in the linked chunk.
+        assert!(result.request_pagination.not());
    }

    #[test]
@@ -918,9 +954,11 @@ mod tests {
            active_receipt,
            with_threading_support,
        );
-        assert_eq!(result.unwrap(), event_id!("$2"));
+        assert_eq!(result.receipt.unwrap(), event_id!("$2"));
        // And there are no pending receipts.
        assert!(pending_receipts.is_empty());
+        // We don't need to paginate, as we found a receipt in the linked chunk.
+        assert!(result.request_pagination.not());
    }

    #[test]
@@ -960,9 +998,11 @@ mod tests {
            active_receipt,
            with_threading_support,
        );
-        assert_eq!(result.unwrap(), event_id!("$2"));
+        assert_eq!(result.receipt.unwrap(), event_id!("$2"));
        // And there are no pending receipts.
        assert!(pending_receipts.is_empty());
+        // We don't need to paginate, as we found a receipt in the linked chunk.
+        assert!(result.request_pagination.not());
    }

    #[test]
@@ -1003,10 +1043,13 @@ mod tests {
            with_threading_support,
        );

-        assert!(result.is_none());
+        assert!(result.receipt.is_none());
        // And there's a new pending receipt for $4.
        assert_eq!(pending_receipts.len(), 1);
        assert_eq!(pending_receipts.get(0).unwrap(), event_id!("$4"));
+        // We would need to paginate, as we haven't found a better receipt in the linked
+        // chunk.
+        assert!(result.request_pagination);
    }

    #[test]
@@ -1043,9 +1086,11 @@ mod tests {
            active_receipt,
            with_threading_support,
        );
-        assert_eq!(result.unwrap(), event_id!("$2"));
+        assert_eq!(result.receipt.unwrap(), event_id!("$2"));
        // And there are no more pending receipts.
        assert!(pending_receipts.is_empty());
+        // We don't need to paginate, as we found a receipt in the linked chunk.
+        assert!(result.request_pagination.not());
    }

    #[test]
@@ -1093,12 +1138,15 @@ mod tests {
            active_receipt,
            with_threading_support,
        );
-        assert_eq!(result.unwrap(), event_id!("$4"));
+        assert_eq!(result.receipt.unwrap(), event_id!("$4"));

        // Receipt 6 is still pending, and there's a new pending receipt for 7 too. ($2
        // has been cleaned because it has been seen).
        assert_eq!(pending_receipts.len(), 2);
        assert!(pending_receipts.iter().any(|ev| ev == event_id!("$6")));
        assert!(pending_receipts.iter().any(|ev| ev == event_id!("$7")));
+
+        // We don't need to paginate, as we found a receipt in the linked chunk.
+        assert!(result.request_pagination.not());
    }
 }
@@ -71,7 +71,13 @@ use super::{
    RoomEventCacheLinkedChunkUpdate, RoomEventCacheUpdate, RoomEventCacheUpdateSender,
    sort_positions_descending,
 };
-use crate::{Room, event_cache::caches::pagination::SharedPaginationStatus, room::WeakRoom};
+use crate::{
+    Room,
+    event_cache::{
+        automatic_pagination::AutomaticPagination, caches::pagination::SharedPaginationStatus,
+    },
+    room::WeakRoom,
+};

 /// Key for the event-focused caches.
 #[derive(Hash, PartialEq, Eq)]
@@ -141,6 +147,9 @@ pub struct RoomEventCacheState {
    /// An atomic count of the current number of subscriber of the
    /// [`super::RoomEventCache`].
    subscriber_count: Arc<AtomicUsize>,
+
+    /// A copy of the automatic pagination API object.
+    automatic_pagination: Option<AutomaticPagination>,
 }

 impl RoomEventCacheState {
@@ -301,6 +310,7 @@ impl LockedRoomEventCacheState {
        linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
        store: EventCacheStoreLock,
        pagination_status: SharedObservable<SharedPaginationStatus>,
+        automatic_pagination: Option<AutomaticPagination>,
    ) -> Result<Self, EventCacheError> {
        let store_guard = match store.lock().await? {
            // Lock is clean: all good!
@@ -383,6 +393,7 @@ impl LockedRoomEventCacheState {
            waited_for_initial_prev_token: false,
            subscriber_count: Default::default(),
            pinned_event_cache: OnceLock::new(),
+            automatic_pagination,
        }))
    }
 }
@@ -1025,9 +1036,8 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
            return Ok(());
        };

-        let state = &mut *self.state;
-        let user_id = &state.own_user_id;
-        let room_id = &state.room_id;
+        let user_id = &self.state.own_user_id;
+        let room_id = &self.state.room_id;

        let prev_read_receipts = room.read_receipts().clone();
        let mut read_receipts = prev_read_receipts.clone();
@@ -1036,9 +1046,10 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
            user_id,
            room_id,
            receipt_event,
-            &state.room_linked_chunk,
+            &self.state.room_linked_chunk,
            &mut read_receipts,
-            state.enabled_thread_support,
+            self.state.enabled_thread_support,
+            self.state.automatic_pagination.as_ref(),
        );

        if prev_read_receipts != read_receipts {
@@ -30,8 +30,7 @@
 use std::{
    collections::HashMap,
    fmt,
-    ops::{Deref, DerefMut},
-    sync::{Arc, OnceLock},
+    sync::{Arc, OnceLock, RwLock as StdRwLock, RwLockReadGuard, RwLockWriteGuard},
 };

 use futures_util::future::try_join_all;
@@ -57,6 +56,7 @@ use crate::{
    paginators::PaginatorError,
 };

+mod automatic_pagination;
 mod caches;
 mod deduplicator;
 mod persistence;
@@ -78,6 +78,8 @@ pub use caches::{
 #[cfg(feature = "e2e-encryption")]
 pub use redecryptor::{DecryptionRetryRequest, RedecryptorReport};

+pub use crate::event_cache::automatic_pagination::AutomaticPagination;
+
 /// An error observed in the [`EventCache`].
 #[derive(thiserror::Error, Clone, Debug)]
 pub enum EventCacheError {
@@ -214,7 +216,7 @@ impl EventCache {
        Self {
            inner: Arc::new(EventCacheInner {
                client: weak_client,
-                config: RwLock::new(EventCacheConfig::default()),
+                config: StdRwLock::new(EventCacheConfig::default()),
                store: event_cache_store,
                multiple_room_updates_lock: Default::default(),
                by_room: Default::default(),
@@ -224,6 +226,7 @@ impl EventCache {
                linked_chunk_update_sender,
                #[cfg(feature = "e2e-encryption")]
                redecryption_channels,
+                automatic_pagination: OnceLock::new(),
                thread_subscriber_sender,
            }),
        }
@@ -231,13 +234,13 @@ impl EventCache {

    /// Get a read-only handle to the global configuration of the
    /// [`EventCache`].
-    pub async fn config(&self) -> impl Deref<Target = EventCacheConfig> + '_ {
-        self.inner.config.read().await
+    pub fn config(&self) -> RwLockReadGuard<'_, EventCacheConfig> {
+        self.inner.config.read().unwrap()
    }

    /// Get a writable handle to the global configuration of the [`EventCache`].
-    pub async fn config_mut(&self) -> impl DerefMut<Target = EventCacheConfig> + '_ {
-        self.inner.config.write().await
+    pub fn config_mut(&self) -> RwLockWriteGuard<'_, EventCacheConfig> {
+        self.inner.config.write().unwrap()
    }

    /// Subscribes to updates that a thread subscription has been sent.
@@ -318,6 +321,15 @@ 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));
+            } else {
+                trace!("automatic paginations API is disabled");
+            }
+
            Arc::new(EventCacheDropHandles {
                _listen_updates_task: listen_updates_task,
                _ignore_user_list_update_task: ignore_user_list_update_task,
@@ -377,6 +389,13 @@ impl EventCache {
    pub fn subscribe_to_room_generic_updates(&self) -> Receiver<RoomEventCacheGenericUpdate> {
        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<AutomaticPagination> {
+        self.inner.automatic_pagination.get().cloned()
+    }
 }

 /// Global configuration for the [`EventCache`], applied to every single room.
@@ -387,15 +406,46 @@ pub struct EventCacheConfig {

    /// Maximum number of pinned events to load, for any room.
    pub max_pinned_events_to_load: usize,
+
+    /// Whether to automatically backpaginate a room under certain conditions.
+    ///
+    /// Off by default.
+    pub experimental_auto_backpagination: bool,
+
+    /// The maximum number of allowed room paginations, for a given room, that
+    /// can be executed in the automatic paginations task.
+    ///
+    /// 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,
 }

 impl EventCacheConfig {
    /// The default maximum number of pinned events to load.
-    const DEFAULT_MAX_EVENTS_TO_LOAD: usize = 128;
+    pub const DEFAULT_MAX_EVENTS_TO_LOAD: usize = 128;

    /// The default maximum number of concurrent requests to perform when
    /// loading the pinned events.
-    const DEFAULT_MAX_CONCURRENT_REQUESTS: usize = 8;
+    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;
 }

 impl Default for EventCacheConfig {
@@ -403,6 +453,9 @@ 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,
        }
    }
 }
@@ -415,7 +468,7 @@ struct EventCacheInner {
    client: WeakClient,

    /// Global configuration for the event cache.
-    config: RwLock<EventCacheConfig>,
+    config: StdRwLock<EventCacheConfig>,

    /// Reference to the underlying store.
    store: EventCacheStoreLock,
@@ -441,6 +494,9 @@ struct EventCacheInner {
    /// Needs to live here, so it may be passed to each [`RoomEventCache`]
    /// instance.
    ///
+    /// It's a `OnceLock` because its initialization is deferred to
+    /// [`EventCache::subscribe`].
+    ///
    /// See doc comment of [`EventCache::auto_shrink_linked_chunk_task`].
    auto_shrink_sender: OnceLock<mpsc::Sender<AutoShrinkChannelPayload>>,

@@ -468,6 +524,12 @@ struct EventCacheInner {

    #[cfg(feature = "e2e-encryption")]
    redecryption_channels: redecryptor::RedecryptorChannels,
+
+    /// State for the automatic pagination mechanism.
+    ///
+    /// Depends on the [`EventCacheConfig::experimental_auto_backpagination`]
+    /// flag to be set at subscription time.
+    automatic_pagination: OnceLock<AutomaticPagination>,
 }

 type AutoShrinkChannelPayload = OwnedRoomId;
@@ -627,6 +689,7 @@ impl EventCacheInner {
                        "we must have called `EventCache::subscribe()` before calling here.",
                    ),
                    self.store.clone(),
+                    self.automatic_pagination.get().cloned(),
                )
                .await?;

@@ -602,7 +602,7 @@ async fn test_reset_while_backpaginating() {
    wait_for_initial_events(events, &mut room_stream).await;

    // We're going to cause a small race:
-    // - a background request to sync will be sent,
+    // - a pagination request to sync will be sent,
    // - a backpagination will be sent concurrently.
    //
    // So events have to happen in this order:
@@ -24,9 +24,12 @@
 //! This avoids potential race conditions where a sync could be done, but the
 //! processing by the event cache isn't, at the time we check the unread counts.

+use std::time::Duration;
+
 use matrix_sdk::{
-    ThreadingSupport, assert_let_timeout, event_cache::RoomEventCacheUpdate,
-    test_utils::mocks::MatrixMockServer,
+    ThreadingSupport, assert_let_timeout,
+    event_cache::RoomEventCacheUpdate,
+    test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate},
 };
 use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory};
 use ruma::{
@@ -715,3 +718,77 @@ async fn test_unread_counts_updated_after_duplicate_only_sync_response() {
    // The message counts are properly updated (zero new message unread after $2).
    assert_eq!(room.num_unread_messages(), 0);
 }
+
+/// Test that the unread count computation causes a back-pagination, when it
+/// can't find an event pointed to by a read receipt.
+#[async_test]
+async fn test_compute_unread_counts_triggers_backpaginations() {
+    let server = MatrixMockServer::new().await;
+    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().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();
+    let (_, mut room_cache_updates) = room_event_cache.subscribe().await.unwrap();
+    assert!(room_cache_updates.is_empty());
+
+    // Already set up the mock for /messages, as the background pagination will hit
+    // it as soon as the sync is received.
+    server
+        .mock_room_messages()
+        .match_from("prev_batch")
+        .ok(RoomMessagesResponseTemplate::default()
+            .events(vec![
+                f.text_msg("hello 2").event_id(event_id!("$2")),
+                f.text_msg("hello 1").event_id(event_id!("$1")),
+            ])
+            .with_delay(Duration::from_millis(100)))
+        .mock_once()
+        .mount()
+        .await;
+
+    // Starting with a gappy sync, with a room with two messages from Bob, and a
+    // receipt on Bob's first message $1, which is missing from the timeline,
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("hello 3").event_id(event_id!("$3")))
+                .add_timeline_event(f.text_msg("hello 4").event_id(event_id!("$4")))
+                .add_receipt(
+                    f.read_receipts()
+                        .add(
+                            event_id!("$1"),
+                            own_user_id,
+                            ReceiptType::Read,
+                            ReceiptThread::Unthreaded,
+                        )
+                        .into_event(),
+                )
+                .set_timeline_limited()
+                .set_timeline_prev_batch("prev_batch".to_owned()),
+        )
+        .await;
+
+    // (one vector diff update, one read receipt update)
+    assert_let_timeout!(Ok(_) = room_cache_updates.recv());
+    assert_let_timeout!(Ok(_) = room_cache_updates.recv());
+
+    // The message counts are properly updated (two unread messages in the
+    // timeline, which are $3 and $4).
+    assert_eq!(room.num_unread_messages(), 2);
+
+    // Then, there's a background pagination happening in the room, which will fetch
+    // the missing $1 and $2.
+    assert_let_timeout!(Duration::from_millis(150), Ok(_) = room_cache_updates.recv());
+
+    // The message counts are properly updated (three messages after $1).
+    assert_eq!(room.num_unread_messages(), 3);
+}
@@ -153,6 +153,7 @@ async fn main() -> Result<()> {
    });

    let event_cache = client.event_cache();
+    event_cache.config_mut().experimental_auto_backpagination = true;
    event_cache.subscribe()?;

    let terminal = ratatui::init();