Commit 5ec7600f0a5f
tarball1 file changed+181-0
@@ -483,3 +483,184 @@ async fn test_segmented_history_preserves_sync_arrivals_during_a_fill() {
483483
vec![event_id!("$old"), event_id!("$middle"), event_id!("$new"), event_id!("$live")]484484
);485485
}486
+487
+async fn wait_for_messages_request(server: &MatrixMockServer, count: usize) {488
+ tokio::time::timeout(Duration::from_secs(8), async {489
+ while server490
+ .received_requests()491
+ .await492
+ .unwrap()493
+ .iter()494
+ .filter(|request| request.url.path().ends_with("/messages"))495
+ .count()496
+ < count497
+ {498
+ tokio::task::yield_now().await;499
+ }500
+ })501
+ .await502
+ .unwrap();503
+}504
+505
+#[async_test]506
+async fn test_history_commit_fences_an_inflight_pagination() {507
+ let server = MatrixMockServer::new().await;508
+ let client = server.client_builder().build().await;509
+ client.event_cache().subscribe().unwrap();510
+ let room_id = room_id!("!history:example.org");511
+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));512
+ let room = server513
+ .sync_room(514
+ &client,515
+ JoinedRoomBuilder::new(room_id)516
+ .add_timeline_event(f.text_msg("old").event_id(event_id!("$old")))517
+ .set_timeline_prev_batch("older"),518
+ )519
+ .await;520
+ let (cache, _handles) = room.event_cache().await.unwrap();521
+ server522
+ .mock_room_messages()523
+ .match_from("older")524
+ .ok(RoomMessagesResponseTemplate::default()525
+ .events(vec![f.text_msg("stale").event_id(event_id!("$stale"))])526
+ .with_delay(Duration::from_millis(300)))527
+ .mock_once()528
+ .mount()529
+ .await;530
+ let pagination = tokio::spawn({531
+ let cache = cache.clone();532
+ async move { cache.pagination().run_backwards_once(10).await }533
+ });534
+ wait_for_messages_request(&server, 1).await;535
+ server536
+ .mock_room_messages()537
+ .match_from("older")538
+ .ok(RoomMessagesResponseTemplate::default()539
+ .events(vec![f.text_msg("fresh").event_id(event_id!("$fresh"))]))540
+ .mock_once()541
+ .mount()542
+ .await;543
+544
+ cache545
+ .append_history_section(546
+ vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],547
+ Some("new-edge".into()),548
+ )549
+ .await550
+ .unwrap();551
+ tokio::time::timeout(Duration::from_secs(3), pagination).await.unwrap().unwrap().unwrap();552
+553
+ let ids: Vec<_> =554
+ cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();555
+ assert_eq!(ids, vec![event_id!("$fresh"), event_id!("$old"), event_id!("$new")]);556
+}557
+558
+#[async_test]559
+async fn test_history_commit_leaves_no_stale_post_processing_diffs() {560
+ let server = MatrixMockServer::new().await;561
+ let client = server.client_builder().build().await;562
+ client.event_cache().subscribe().unwrap();563
+ let room_id = room_id!("!history:example.org");564
+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));565
+ let room = server566
+ .sync_room(567
+ &client,568
+ JoinedRoomBuilder::new(room_id)569
+ .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),570
+ )571
+ .await;572
+ let (cache, _handles) = room.event_cache().await.unwrap();573
+ let (_, mut updates) = cache.subscribe().await.unwrap();574
+ let secret = event_id!("$secret");575
+576
+ cache577
+ .append_history_section(578
+ vec![579
+ f.text_msg("secret").event_id(secret).into_event(),580
+ f.redaction(secret).event_id(event_id!("$redaction")).into_event(),581
+ ],582
+ Some("section-edge".into()),583
+ )584
+ .await585
+ .unwrap();586
+ tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();587
+588
+ let live = event_id!("$live");589
+ server590
+ .sync_room(591
+ &client,592
+ JoinedRoomBuilder::new(room_id).add_timeline_event(f.text_msg("live").event_id(live)),593
+ )594
+ .await;595
+ let update =596
+ tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();597
+ let matrix_sdk::event_cache::RoomEventCacheUpdate::UpdateTimelineEvents(update) = update else {598
+ panic!("expected timeline update")599
+ };600
+ for diff in update.diffs {601
+ match diff {602
+ eyeball_im::VectorDiff::Append { values } => {603
+ assert!(values.iter().all(|event| event.event_id().as_deref() == Some(live)))604
+ }605
+ eyeball_im::VectorDiff::PushBack { value } => {606
+ assert_eq!(value.event_id().as_deref(), Some(live))607
+ }608
+ other => panic!("unexpected diff after the commit {other:?}"),609
+ }610
+ }611
+}612
+613
+#[async_test]614
+async fn test_sync_superseding_a_trailing_gap_stops_its_pagination_cleanly() {615
+ let server = MatrixMockServer::new().await;616
+ let client = server.client_builder().build().await;617
+ client.event_cache().subscribe().unwrap();618
+ let room_id = room_id!("!history:example.org");619
+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));620
+ let room = server621
+ .sync_room(622
+ &client,623
+ JoinedRoomBuilder::new(room_id)624
+ .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),625
+ )626
+ .await;627
+ server628
+ .sync_room(629
+ &client,630
+ JoinedRoomBuilder::new(room_id).set_timeline_limited().set_timeline_prev_batch("trail"),631
+ )632
+ .await;633
+ let (cache, _handles) = room.event_cache().await.unwrap();634
+ server635
+ .mock_room_messages()636
+ .match_from("trail")637
+ .ok(RoomMessagesResponseTemplate::default()638
+ .events(vec![f.text_msg("stale").event_id(event_id!("$stale"))])639
+ .with_delay(Duration::from_millis(300)))640
+ .mock_once()641
+ .mount()642
+ .await;643
+ let pagination = tokio::spawn({644
+ let cache = cache.clone();645
+ async move { cache.pagination().run_backwards_once(10).await }646
+ });647
+ wait_for_messages_request(&server, 1).await;648
+649
+ server650
+ .sync_room(651
+ &client,652
+ JoinedRoomBuilder::new(room_id)653
+ .add_timeline_event(f.text_msg("new").event_id(event_id!("$new")))654
+ .set_timeline_limited()655
+ .set_timeline_prev_batch("newer"),656
+ )657
+ .await;658
+ tokio::time::timeout(Duration::from_secs(3), pagination).await.unwrap().unwrap().unwrap();659
+660
+ let ids: Vec<_> =661
+ cache.history_snapshot().await.unwrap().0.iter().filter_map(|e| e.event_id()).collect();662
+ assert!(!ids.contains(&event_id!("$stale").to_owned()));663
+ let gaps = cache.history_gaps().await.unwrap();664
+ assert_eq!(gaps.len(), 1);665
+ assert_eq!(gaps[0].older_boundary.as_deref(), Some(event_id!("$old")));666
+}