integration test: fix racy behavior in the test_toggling_reaction

The room sync could be over, and then the timeline items would not contain the updated item yet, if the timeline didn't get a chance to finish its
internal update. Fix this by:

1. limiting the sync time to 10 seconds
2. giving up to one second for the timeline to update internally, by watching its stream of events (we're ignoring stream updates just after this loop)

Author: bnjbvr

testing/matrix-sdk-integration-testing/src/tests/reactions.rs

@@ -14,7 +14,7 @@
 
 use std::{sync::Arc, time::Duration};
 
-use anyhow::{Ok, Result};
+use anyhow::Result;
 use assert_matches::assert_matches;
 use assert_matches2::assert_let;
 use assign::assign;
@@ -35,16 +35,34 @@ use tokio::time::timeout;
 
 use crate::helpers::TestClientBuilder;
 
+/// Sync until we receive an update for a room.
+///
+/// Beware: it may sync forever, if it doesn't receive such an update!
+async fn sync_until_update_for_room(client: &Client, room_id: &RoomId) -> Result<()> {
+    client
+        .sync_with_callback(SyncSettings::default(), |response| async move {
+            if response.rooms.join.iter().any(|(id, _)| id == room_id) {
+                LoopCtrl::Break
+            } else {
+                LoopCtrl::Continue
+            }
+        })
+        .await?;
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 async fn test_toggling_reaction() -> Result<()> {
     // Set up sync for user Alice, and create a room.
     let alice = TestClientBuilder::new("alice".to_owned()).use_sqlite().build().await?;
+
     let user_id = alice.user_id().unwrap().to_owned();
     let room = alice
         .create_room(assign!(CreateRoomRequest::new(), {
             is_direct: true,
         }))
         .await?;
+
     let room_id = room.room_id();
 
     // Create a timeline for this room.
@@ -57,12 +75,25 @@ async fn test_toggling_reaction() -> Result<()> {
     // Sync until the remote echo arrives.
     let mut num_attempts = 0;
     let event_id = loop {
-        sync_room(&alice, room_id).await?;
+        // We expect a quick update from sync, so timeout after 10 seconds, and ignore
+        // timeouts; they might just indicate we received the necessary
+        // information on the previous attempt, but racily tried to read the
+        // timeline items.
+        if let Ok(sync_result) =
+            timeout(Duration::from_secs(10), sync_until_update_for_room(&alice, room_id)).await
+        {
+            sync_result?;
+        }
+
+        // Let time to the timeline for processing the update, at most 1 second.
+        let _ = timeout(Duration::from_secs(1), stream.next()).await;
+
         let items = timeline.items().await;
         let last_item = items.last().unwrap().as_event().unwrap();
         if !last_item.is_local_echo() {
             break last_item.event_id().unwrap().to_owned();
         }
+
         if num_attempts == 2 {
             panic!("had 3 sync responses and no echo of our own event");
         }
@@ -181,19 +212,6 @@ async fn assert_remote_added(
     assert!(reaction.unwrap().timestamp <= MilliSecondsSinceUnixEpoch::now());
 }
 
-async fn sync_room(client: &Client, room_id: &RoomId) -> Result<()> {
-    client
-        .sync_with_callback(SyncSettings::default(), |response| async move {
-            if response.rooms.join.iter().any(|(id, _)| id == room_id) {
-                LoopCtrl::Break
-            } else {
-                LoopCtrl::Continue
-            }
-        })
-        .await?;
-    Ok(())
-}
-
 async fn assert_event_is_updated(
     stream: &mut (impl Stream<Item = VectorDiff<Arc<TimelineItem>>> + Unpin),
     event_id: &EventId,