sdk-ui: fix max concurrent requests for pinned events timeline.

`RequestConfig::max_concurrent_requests` won't work for requests to a same endpoint, since it's a global value that can only be passed to `HttpClient` when created.

To replace it I added a buffered stream that can run at most `MAX_PINNED_EVENTS_CONCURRENT_REQUESTS` (10) requests concurrently.

There is also a test to make sure this max concurrency mechanism works.

Author: jmartinesp

crates/matrix-sdk-ui/src/timeline/mod.rs

@@ -101,6 +101,7 @@ pub use self::{
     inner::default_event_filter,
     item::{TimelineItem, TimelineItemKind},
     pagination::LiveBackPaginationStatus,
+    pinned_events_loader::MAX_PINNED_EVENTS_CONCURRENT_REQUESTS,
     polls::PollResult,
     traits::RoomExt,
     virtual_item::VirtualTimelineItem,

crates/matrix-sdk-ui/src/timeline/pinned_events_loader.rs

@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{fmt::Formatter, num::NonZeroUsize, sync::Arc};
+use std::{fmt::Formatter, sync::Arc};
 
-use futures_util::future::join_all;
+use futures_util::{stream, StreamExt};
 use matrix_sdk::{
     config::RequestConfig, event_cache::paginator::PaginatorError, Room, SendOutsideWasm,
     SyncOutsideWasm,
@@ -26,7 +26,7 @@ use tracing::{debug, warn};
 
 use crate::timeline::event_handler::TimelineEventKind;
 
-const MAX_CONCURRENT_REQUESTS: usize = 10;
+pub const MAX_PINNED_EVENTS_CONCURRENT_REQUESTS: usize = 10;
 
 /// Utility to load the pinned events in a room.
 pub struct PinnedEventsLoader {
@@ -66,37 +66,33 @@ impl PinnedEventsLoader {
             return Ok(Vec::new());
         }
 
-        let request_config = Some(
-            RequestConfig::default()
-                .retry_limit(3)
-                .max_concurrent_requests(NonZeroUsize::new(MAX_CONCURRENT_REQUESTS)),
-        );
-
-        let new_events = join_all(pinned_event_ids.into_iter().map(|event_id| {
-            let provider = self.room.clone();
-            async move {
-                match provider.load_event_with_relations(&event_id, request_config).await {
-                    Ok((event, related_events)) => {
-                        let mut events = vec![event];
-                        events.extend(related_events);
-                        Some(events)
-                    }
-                    Err(err) => {
-                        warn!("error when loading pinned event: {err}");
-                        None
+        let request_config = Some(RequestConfig::default().retry_limit(3));
+
+        let mut loaded_events: Vec<SyncTimelineEvent> =
+            stream::iter(pinned_event_ids.into_iter().map(|event_id| {
+                let provider = self.room.clone();
+                async move {
+                    match provider.load_event_with_relations(&event_id, request_config).await {
+                        Ok((event, related_events)) => {
+                            let mut events = vec![event];
+                            events.extend(related_events);
+                            Some(events)
+                        }
+                        Err(err) => {
+                            warn!("error when loading pinned event: {err}");
+                            None
+                        }
                     }
                 }
-            }
-        }))
-        .await;
-
-        let mut loaded_events = new_events
-            .into_iter()
+            }))
+            .buffer_unordered(MAX_PINNED_EVENTS_CONCURRENT_REQUESTS)
             // Get only the `Some<Vec<_>>` results
-            .flatten()
+            .flat_map(stream::iter)
             // Flatten the `Vec`s into a single one containing all their items
-            .flatten()
-            .collect::<Vec<SyncTimelineEvent>>();
+            .flat_map(stream::iter)
+            .collect()
+            .await;
+
         if loaded_events.is_empty() {
             return Err(PinnedEventsLoaderError::TimelineReloadFailed);
         }

crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs

@@ -12,13 +12,19 @@ use matrix_sdk::{
 use matrix_sdk_base::deserialized_responses::TimelineEvent;
 use matrix_sdk_test::{async_test, JoinedRoomBuilder, StateTestEvent, SyncResponseBuilder, BOB};
 use matrix_sdk_ui::{
-    timeline::{TimelineFocus, TimelineItemContent},
+    timeline::{
+        RoomExt, TimelineFocus, TimelineItemContent, MAX_PINNED_EVENTS_CONCURRENT_REQUESTS,
+    },
     Timeline,
 };
 use ruma::{event_id, owned_room_id, MilliSecondsSinceUnixEpoch, OwnedRoomId};
 use serde_json::json;
 use stream_assert::assert_pending;
-use wiremock::MockServer;
+use tokio::time::sleep;
+use wiremock::{
+    matchers::{header, method, path_regex},
+    Mock, MockServer, ResponseTemplate,
+};
 
 use crate::{mock_event, mock_sync};
 
@@ -624,6 +630,78 @@ async fn test_edited_events_survive_pinned_event_ids_change() {
     assert_pending!(timeline_stream);
 }
 
+#[async_test]
+async fn test_ensure_max_concurrency_is_observed() {
+    let (client, server) = logged_in_client_with_server().await;
+    let room_id = owned_room_id!("!a_room:example.org");
+
+    let pinned_event_ids: Vec<String> = (0..100).map(|idx| format!("${idx}")).collect();
+
+    let joined_room_builder = JoinedRoomBuilder::new(&room_id)
+        // Set up encryption
+        .add_state_event(StateTestEvent::Encryption)
+        // Add 100 pinned events
+        .add_state_event(StateTestEvent::Custom(json!(
+            {
+                "content": {
+                    "pinned": pinned_event_ids
+                },
+                "event_id": "$15139375513VdeRF:localhost",
+                "origin_server_ts": 151393755,
+                "sender": "@example:localhost",
+                "state_key": "",
+                "type": "m.room.pinned_events",
+                "unsigned": {
+                    "age": 703422
+                }
+            }
+        )));
+
+    // Amount of time to delay the response of an /event mock request, in ms.
+    let request_delay = 50;
+    let pinned_event =
+        EventFactory::new().room(&room_id).sender(*BOB).text_msg("A message").into_raw_timeline();
+    Mock::given(method("GET"))
+        .and(path_regex(r"/_matrix/client/r0/rooms/.*/event/.*"))
+        .and(header("authorization", "Bearer 1234"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_delay(Duration::from_millis(request_delay))
+                .set_body_json(pinned_event.json()),
+        )
+        // Verify this endpoint is only called the max concurrent amount of times.
+        .expect(MAX_PINNED_EVENTS_CONCURRENT_REQUESTS as u64)
+        .mount(&server)
+        .await;
+
+    let mut sync_response_builder = SyncResponseBuilder::new();
+    let sync_settings = SyncSettings::new().timeout(Duration::from_millis(3000));
+    let json_response =
+        sync_response_builder.add_joined_room(joined_room_builder).build_json_sync_response();
+    mock_sync(&server, json_response, None).await;
+    let _ = client.sync_once(sync_settings.clone()).await;
+
+    let room = client.get_room(&room_id).unwrap();
+
+    // Start loading the pinned event timeline asynchronously.
+    let handle = tokio::spawn({
+        let timeline_builder = room
+            .timeline_builder()
+            .with_focus(TimelineFocus::PinnedEvents { max_events_to_load: 100 });
+        async {
+            let _ = timeline_builder.build().await;
+        }
+    });
+
+    // Give it time to load events. As each request takes `request_delay`, we should
+    // have exactly `MAX_PINNED_EVENTS_CONCURRENT_REQUESTS` if the max
+    // concurrency setting is honoured.
+    sleep(Duration::from_millis(request_delay)).await;
+
+    // Abort handle to stop requests from being processed.
+    handle.abort();
+}
+
 struct TestHelper {
     pub client: Client,
     pub server: MockServer,