macltok

public

Star0

Watch0

45 commits 3 branches 0 tags 0 releases3.0 MiB

#![cfg(feature = "mirror")]

use std::time::Duration;

use futures_util::StreamExt;
use macltok::{
    ClientHandle, Command, Engine, EngineConfig, Event, Mirror, MirrorHandle, StateChange,
    Subscription, TimelineChange, TimelineCommand, TimelineDiff, TimelineId, TimelineItem, Update,
    UploadOutcome,
};
use matrix_sdk::{
    ruma::{
        OwnedRoomId, OwnedTransactionId, TransactionId, api::client::sync::sync_events::v5,
        event_id, room_id,
    },
    test_utils::mocks::MatrixMockServer,
};
use tokio::sync::mpsc;

struct Harness {
    handle: ClientHandle,
    mirror: MirrorHandle,
    inject: mpsc::UnboundedSender<Event>,
    room_id: OwnedRoomId,
    _server: MatrixMockServer,
}

/// the mirror reads the real engine's stream with synthetic events mixed in, so a test can
/// hand it a diff the engine never would
async fn start(config: EngineConfig, capacity: usize) -> Harness {
    let server = MatrixMockServer::new().await;
    server
        .mock_sliding_sync()
        .ok(v5::Response::new("1".to_owned()))
        .mount()
        .await;
    let client = server.client_builder().build().await;
    let room_id = room_id!("!mirror:localhost");
    server.mock_room_state_encryption().plain().mount().await;
    server.sync_joined_room(&client, room_id).await;
    let (engine, handle, events) =
        Engine::from_client(client, config.with_initial_history_limit(0))
            .await
            .unwrap();
    tokio::spawn(engine.run());
    let (inject, mut injected) = mpsc::unbounded_channel();
    let injected = futures_util::stream::poll_fn(move |context| injected.poll_recv(context));
    let (mirror, mirror_handle) = Mirror::with_capacity(
        handle.clone(),
        futures_util::stream::select(events, injected),
        capacity,
    );
    tokio::spawn(mirror.run());
    Harness {
        handle,
        mirror: mirror_handle,
        inject,
        room_id: room_id.to_owned(),
        _server: server,
    }
}

async fn next_matching(
    subscription: &mut Subscription,
    predicate: impl Fn(&StateChange) -> bool,
) -> Update {
    tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            let update = subscription.next().await.expect("the mirror stopped");
            if predicate(&update.change) {
                return update;
            }
        }
    })
    .await
    .expect("expected update did not arrive")
}

fn timeline_change(
    change: &StateChange,
    id: &TimelineId,
    predicate: impl Fn(&TimelineChange) -> bool,
) -> bool {
    matches!(change, StateChange::Timeline { timeline_id, change } if timeline_id == id && predicate(change))
}

fn item(timestamp_ms: u64) -> TimelineItem {
    TimelineItem::DayDivider { timestamp_ms }
}

fn reset(timeline_id: &TimelineId, items: Vec<TimelineItem>) -> Event {
    Event::TimelineReset {
        timeline_id: timeline_id.clone(),
        items: items.into(),
    }
}

/// the room never opens, so every resync is refused and the divergence holds still to be looked at
#[tokio::test]
async fn diverged_batch_keeps_last_good_items_until_a_reset() {
    let harness = start(EngineConfig::default(), 256).await;
    let timeline_id = TimelineId::from(harness.room_id.clone());
    let mut updates = harness.mirror.subscribe();
    harness
        .inject
        .send(reset(&timeline_id, vec![item(1), item(2)]))
        .unwrap();
    for diffs in [
        vec![
            TimelineDiff::PushBack { value: item(3) },
            TimelineDiff::Remove { index: 7 },
        ],
        vec![TimelineDiff::PushBack { value: item(4) }],
    ] {
        harness
            .inject
            .send(Event::TimelineChanged {
                timeline_id: timeline_id.clone(),
                diffs,
            })
            .unwrap();
    }
    next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |change| {
            matches!(change, TimelineChange::Diverged)
        })
    })
    .await;
    let snapshot = harness.mirror.snapshot();
    let timeline = &snapshot.timelines[&timeline_id];
    assert!(timeline.diverged);
    assert_eq!(timeline.items, [item(1), item(2)].into_iter().collect());

    harness
        .inject
        .send(reset(&timeline_id, vec![item(5)]))
        .unwrap();
    let update = next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |_| true)
    })
    .await;
    assert!(
        timeline_change(&update.change, &timeline_id, |change| {
            matches!(change, TimelineChange::Reset { .. })
        }),
        "a diverged timeline took {update:?}"
    );
    let snapshot = harness.mirror.snapshot();
    let timeline = &snapshot.timelines[&timeline_id];
    assert!(!timeline.diverged);
    assert_eq!(timeline.items, [item(5)].into_iter().collect());
}

#[tokio::test]
async fn a_diff_for_a_timeline_without_a_reset_diverges_it() {
    let harness = start(EngineConfig::default(), 256).await;
    let timeline_id = TimelineId::from(harness.room_id.clone());
    let mut updates = harness.mirror.subscribe();
    harness
        .inject
        .send(Event::TimelineChanged {
            timeline_id: timeline_id.clone(),
            diffs: vec![TimelineDiff::PushBack { value: item(1) }],
        })
        .unwrap();
    next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |change| {
            matches!(change, TimelineChange::Diverged)
        })
    })
    .await;
    let snapshot = harness.mirror.snapshot();
    let timeline = &snapshot.timelines[&timeline_id];
    assert!(timeline.diverged);
    assert!(timeline.items.is_empty());
}

#[tokio::test]
async fn room_and_thread_timelines_stay_apart() {
    let harness = start(EngineConfig::default(), 256).await;
    let room = TimelineId::from(harness.room_id.clone());
    let thread = TimelineId::thread(
        harness.room_id.clone(),
        event_id!("$root:localhost").to_owned(),
    );
    let mut updates = harness.mirror.subscribe();
    harness.inject.send(reset(&room, vec![item(1)])).unwrap();
    harness.inject.send(reset(&thread, vec![item(2)])).unwrap();
    harness
        .inject
        .send(Event::TimelineChanged {
            timeline_id: thread.clone(),
            diffs: vec![TimelineDiff::PushBack { value: item(3) }],
        })
        .unwrap();
    next_matching(&mut updates, |change| {
        timeline_change(change, &thread, |change| {
            matches!(change, TimelineChange::Changed { .. })
        })
    })
    .await;
    let snapshot = harness.mirror.snapshot();
    assert_eq!(
        snapshot.timelines[&room].items,
        [item(1)].into_iter().collect()
    );
    assert_eq!(
        snapshot.timelines[&thread].items,
        [item(2), item(3)].into_iter().collect()
    );
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn subscribers_joining_mid_stream_see_no_gap_and_nothing_twice() {
    const CHUNKS: u64 = 64;
    const CHUNK: u64 = 32;
    let harness = start(EngineConfig::default(), 4096).await;
    let transaction_id = TransactionId::new();
    let mut followers = Vec::new();
    for chunk in 0..CHUNKS {
        for uploaded_bytes in chunk * CHUNK + 1..=(chunk + 1) * CHUNK {
            harness
                .inject
                .send(Event::MediaUploadProgress {
                    room_id: harness.room_id.clone(),
                    transaction_id: transaction_id.clone(),
                    uploaded_bytes,
                    total_bytes: CHUNKS * CHUNK,
                    complete: false,
                })
                .unwrap();
        }
        followers.push(tokio::spawn(follow(
            harness.mirror.subscribe(),
            transaction_id.clone(),
            CHUNKS * CHUNK,
        )));
    }
    for follower in followers {
        follower.await.unwrap();
    }
}

async fn follow(mut updates: Subscription, transaction_id: OwnedTransactionId, total: u64) {
    let first = updates.next().await.unwrap();
    let StateChange::State(state) = &first.change else {
        panic!("a subscription opened with {first:?}");
    };
    let mut seen = state
        .uploads
        .get(&transaction_id)
        .map_or(0, |upload| upload.uploaded_bytes);
    let mut sequence = first.sequence;
    while seen < total {
        let update = next_matching(&mut updates, |_| true).await;
        assert_eq!(update.sequence, sequence + 1);
        sequence = update.sequence;
        if let StateChange::Upload { upload, .. } = update.change {
            assert_eq!(upload.uploaded_bytes, seen + 1);
            seen += 1;
        }
    }
}

#[tokio::test]
async fn lagging_subscriber_starts_over_from_a_fresh_state() {
    let harness = start(EngineConfig::default(), 4).await;
    let transaction_id = TransactionId::new();
    let mut updates = harness.mirror.subscribe();
    assert!(matches!(
        updates.next().await.unwrap().change,
        StateChange::State(_)
    ));
    for uploaded_bytes in 1..=20 {
        harness
            .inject
            .send(Event::MediaUploadProgress {
                room_id: harness.room_id.clone(),
                transaction_id: transaction_id.clone(),
                uploaded_bytes,
                total_bytes: 20,
                complete: uploaded_bytes == 20,
            })
            .unwrap();
    }
    tokio::time::timeout(Duration::from_secs(5), async {
        while !harness
            .mirror
            .snapshot()
            .uploads
            .get(&transaction_id)
            .is_some_and(|upload| upload.complete)
        {
            tokio::task::yield_now().await;
        }
    })
    .await
    .expect("the mirror never caught up");

    let update = updates.next().await.unwrap();
    let StateChange::State(state) = &update.change else {
        panic!("a lagged subscription carried on with {update:?}");
    };
    assert!(state.uploads[&transaction_id].complete);
}

#[tokio::test]
async fn resync_does_not_deadlock_on_full_queues() {
    let config = EngineConfig::default()
        .with_command_capacity(1)
        .with_event_capacity(1);
    let harness = start(config, 1024).await;
    let timeline_id = TimelineId::from(harness.room_id.clone());
    let mut updates = harness.mirror.subscribe();
    harness
        .handle
        .open_room(harness.room_id.clone())
        .await
        .unwrap();
    next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |change| {
            matches!(change, TimelineChange::Reset { .. })
        })
    })
    .await;

    let flood = tokio::spawn({
        let handle = harness.handle.clone();
        let timeline_id = timeline_id.clone();
        async move {
            loop {
                let command = Command::Timeline(TimelineCommand::Paginate {
                    timeline_id: timeline_id.clone(),
                    limit: 1,
                });
                if handle.send(command).await.is_err() {
                    return;
                }
            }
        }
    });
    for _ in 0..10 {
        harness
            .inject
            .send(Event::TimelineChanged {
                timeline_id: timeline_id.clone(),
                diffs: vec![TimelineDiff::Remove { index: 1000 }],
            })
            .unwrap();
        next_matching(&mut updates, |change| {
            timeline_change(change, &timeline_id, |change| {
                matches!(change, TimelineChange::Diverged)
            })
        })
        .await;
        next_matching(&mut updates, |change| {
            timeline_change(change, &timeline_id, |change| {
                matches!(change, TimelineChange::Reset { .. })
            })
        })
        .await;
    }
    flood.abort();
}

#[tokio::test]
async fn closing_a_timeline_evicts_it() {
    let harness = start(EngineConfig::default(), 256).await;
    let timeline_id = TimelineId::from(harness.room_id.clone());
    let mut updates = harness.mirror.subscribe();
    harness
        .handle
        .open_room(harness.room_id.clone())
        .await
        .unwrap();
    harness
        .handle
        .close_room(harness.room_id.clone())
        .await
        .unwrap();
    next_matching(&mut updates, |change| {
        timeline_change(change, &timeline_id, |change| {
            matches!(change, TimelineChange::Closed)
        })
    })
    .await;
    assert!(
        !harness
            .mirror
            .snapshot()
            .timelines
            .contains_key(&timeline_id)
    );
}

#[tokio::test]
async fn an_upload_ends_on_finished() {
    let harness = start(EngineConfig::default(), 256).await;
    let transaction_id = TransactionId::new();
    let mut updates = harness.mirror.subscribe();
    harness
        .inject
        .send(Event::MediaUploadProgress {
            room_id: harness.room_id.clone(),
            transaction_id: transaction_id.clone(),
            uploaded_bytes: 5,
            total_bytes: 10,
            complete: false,
        })
        .unwrap();
    next_matching(&mut updates, |change| {
        matches!(change, StateChange::Upload { .. })
    })
    .await;
    assert!(
        harness
            .mirror
            .snapshot()
            .uploads
            .contains_key(&transaction_id)
    );
    harness
        .inject
        .send(Event::MediaUploadFinished {
            room_id: harness.room_id.clone(),
            transaction_id: transaction_id.clone(),
            outcome: UploadOutcome::Cancelled,
        })
        .unwrap();
    next_matching(&mut updates, |change| {
        matches!(change, StateChange::UploadFinished { .. })
    })
    .await;
    assert!(harness.mirror.snapshot().uploads.is_empty());
}

1	#![cfg(feature = "mirror")]
2
3	use std::time::Duration;
4
5	use futures_util::StreamExt;
6	use macltok::{
7	ClientHandle, Command, Engine, EngineConfig, Event, Mirror, MirrorHandle, StateChange,
8	Subscription, TimelineChange, TimelineCommand, TimelineDiff, TimelineId, TimelineItem, Update,
9	UploadOutcome,
10	};
11	use matrix_sdk::{
12	ruma::{
13	OwnedRoomId, OwnedTransactionId, TransactionId, api::client::sync::sync_events::v5,
14	event_id, room_id,
15	},
16	test_utils::mocks::MatrixMockServer,
17	};
18	use tokio::sync::mpsc;
19
20	struct Harness {
21	handle: ClientHandle,
22	mirror: MirrorHandle,
23	inject: mpsc::UnboundedSender<Event>,
24	room_id: OwnedRoomId,
25	_server: MatrixMockServer,
26	}
27
28	/// the mirror reads the real engine's stream with synthetic events mixed in, so a test can
29	/// hand it a diff the engine never would
30	async fn start(config: EngineConfig, capacity: usize) -> Harness {
31	let server = MatrixMockServer::new().await;
32	server
33	.mock_sliding_sync()
34	.ok(v5::Response::new("1".to_owned()))
35	.mount()
36	.await;
37	let client = server.client_builder().build().await;
38	let room_id = room_id!("!mirror:localhost");
39	server.mock_room_state_encryption().plain().mount().await;
40	server.sync_joined_room(&client, room_id).await;
41	let (engine, handle, events) =
42	Engine::from_client(client, config.with_initial_history_limit(0))
43	.await
44	.unwrap();
45	tokio::spawn(engine.run());
46	let (inject, mut injected) = mpsc::unbounded_channel();
47	let injected = futures_util::stream::poll_fn(move \|context\| injected.poll_recv(context));
48	let (mirror, mirror_handle) = Mirror::with_capacity(
49	handle.clone(),
50	futures_util::stream::select(events, injected),
51	capacity,
52	);
53	tokio::spawn(mirror.run());
54	Harness {
55	handle,
56	mirror: mirror_handle,
57	inject,
58	room_id: room_id.to_owned(),
59	_server: server,
60	}
61	}
62
63	async fn next_matching(
64	subscription: &mut Subscription,
65	predicate: impl Fn(&StateChange) -> bool,
66	) -> Update {
67	tokio::time::timeout(Duration::from_secs(5), async {
68	loop {
69	let update = subscription.next().await.expect("the mirror stopped");
70	if predicate(&update.change) {
71	return update;
72	}
73	}
74	})
75	.await
76	.expect("expected update did not arrive")
77	}
78
79	fn timeline_change(
80	change: &StateChange,
81	id: &TimelineId,
82	predicate: impl Fn(&TimelineChange) -> bool,
83	) -> bool {
84	matches!(change, StateChange::Timeline { timeline_id, change } if timeline_id == id && predicate(change))
85	}
86
87	fn item(timestamp_ms: u64) -> TimelineItem {
88	TimelineItem::DayDivider { timestamp_ms }
89	}
90
91	fn reset(timeline_id: &TimelineId, items: Vec<TimelineItem>) -> Event {
92	Event::TimelineReset {
93	timeline_id: timeline_id.clone(),
94	items: items.into(),
95	}
96	}
97
98	/// the room never opens, so every resync is refused and the divergence holds still to be looked at
99	#[tokio::test]
100	async fn diverged_batch_keeps_last_good_items_until_a_reset() {
101	let harness = start(EngineConfig::default(), 256).await;
102	let timeline_id = TimelineId::from(harness.room_id.clone());
103	let mut updates = harness.mirror.subscribe();
104	harness
105	.inject
106	.send(reset(&timeline_id, vec![item(1), item(2)]))
107	.unwrap();
108	for diffs in [
109	vec![
110	TimelineDiff::PushBack { value: item(3) },
111	TimelineDiff::Remove { index: 7 },
112	],
113	vec![TimelineDiff::PushBack { value: item(4) }],
114	] {
115	harness
116	.inject
117	.send(Event::TimelineChanged {
118	timeline_id: timeline_id.clone(),
119	diffs,
120	})
121	.unwrap();
122	}
123	next_matching(&mut updates, \|change\| {
124	timeline_change(change, &timeline_id, \|change\| {
125	matches!(change, TimelineChange::Diverged)
126	})
127	})
128	.await;
129	let snapshot = harness.mirror.snapshot();
130	let timeline = &snapshot.timelines[&timeline_id];
131	assert!(timeline.diverged);
132	assert_eq!(timeline.items, [item(1), item(2)].into_iter().collect());
133
134	harness
135	.inject
136	.send(reset(&timeline_id, vec![item(5)]))
137	.unwrap();
138	let update = next_matching(&mut updates, \|change\| {
139	timeline_change(change, &timeline_id, \|_\| true)
140	})
141	.await;
142	assert!(
143	timeline_change(&update.change, &timeline_id, \|change\| {
144	matches!(change, TimelineChange::Reset { .. })
145	}),
146	"a diverged timeline took {update:?}"
147	);
148	let snapshot = harness.mirror.snapshot();
149	let timeline = &snapshot.timelines[&timeline_id];
150	assert!(!timeline.diverged);
151	assert_eq!(timeline.items, [item(5)].into_iter().collect());
152	}
153
154	#[tokio::test]
155	async fn a_diff_for_a_timeline_without_a_reset_diverges_it() {
156	let harness = start(EngineConfig::default(), 256).await;
157	let timeline_id = TimelineId::from(harness.room_id.clone());
158	let mut updates = harness.mirror.subscribe();
159	harness
160	.inject
161	.send(Event::TimelineChanged {
162	timeline_id: timeline_id.clone(),
163	diffs: vec![TimelineDiff::PushBack { value: item(1) }],
164	})
165	.unwrap();
166	next_matching(&mut updates, \|change\| {
167	timeline_change(change, &timeline_id, \|change\| {
168	matches!(change, TimelineChange::Diverged)
169	})
170	})
171	.await;
172	let snapshot = harness.mirror.snapshot();
173	let timeline = &snapshot.timelines[&timeline_id];
174	assert!(timeline.diverged);
175	assert!(timeline.items.is_empty());
176	}
177
178	#[tokio::test]
179	async fn room_and_thread_timelines_stay_apart() {
180	let harness = start(EngineConfig::default(), 256).await;
181	let room = TimelineId::from(harness.room_id.clone());
182	let thread = TimelineId::thread(
183	harness.room_id.clone(),
184	event_id!("$root:localhost").to_owned(),
185	);
186	let mut updates = harness.mirror.subscribe();
187	harness.inject.send(reset(&room, vec![item(1)])).unwrap();
188	harness.inject.send(reset(&thread, vec![item(2)])).unwrap();
189	harness
190	.inject
191	.send(Event::TimelineChanged {
192	timeline_id: thread.clone(),
193	diffs: vec![TimelineDiff::PushBack { value: item(3) }],
194	})
195	.unwrap();
196	next_matching(&mut updates, \|change\| {
197	timeline_change(change, &thread, \|change\| {
198	matches!(change, TimelineChange::Changed { .. })
199	})
200	})
201	.await;
202	let snapshot = harness.mirror.snapshot();
203	assert_eq!(
204	snapshot.timelines[&room].items,
205	[item(1)].into_iter().collect()
206	);
207	assert_eq!(
208	snapshot.timelines[&thread].items,
209	[item(2), item(3)].into_iter().collect()
210	);
211	}
212
213	#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
214	async fn subscribers_joining_mid_stream_see_no_gap_and_nothing_twice() {
215	const CHUNKS: u64 = 64;
216	const CHUNK: u64 = 32;
217	let harness = start(EngineConfig::default(), 4096).await;
218	let transaction_id = TransactionId::new();
219	let mut followers = Vec::new();
220	for chunk in 0..CHUNKS {
221	for uploaded_bytes in chunk * CHUNK + 1..=(chunk + 1) * CHUNK {
222	harness
223	.inject
224	.send(Event::MediaUploadProgress {
225	room_id: harness.room_id.clone(),
226	transaction_id: transaction_id.clone(),
227	uploaded_bytes,
228	total_bytes: CHUNKS * CHUNK,
229	complete: false,
230	})
231	.unwrap();
232	}
233	followers.push(tokio::spawn(follow(
234	harness.mirror.subscribe(),
235	transaction_id.clone(),
236	CHUNKS * CHUNK,
237	)));
238	}
239	for follower in followers {
240	follower.await.unwrap();
241	}
242	}
243
244	async fn follow(mut updates: Subscription, transaction_id: OwnedTransactionId, total: u64) {
245	let first = updates.next().await.unwrap();
246	let StateChange::State(state) = &first.change else {
247	panic!("a subscription opened with {first:?}");
248	};
249	let mut seen = state
250	.uploads
251	.get(&transaction_id)
252	.map_or(0, \|upload\| upload.uploaded_bytes);
253	let mut sequence = first.sequence;
254	while seen < total {
255	let update = next_matching(&mut updates, \|_\| true).await;
256	assert_eq!(update.sequence, sequence + 1);
257	sequence = update.sequence;
258	if let StateChange::Upload { upload, .. } = update.change {
259	assert_eq!(upload.uploaded_bytes, seen + 1);
260	seen += 1;
261	}
262	}
263	}
264
265	#[tokio::test]
266	async fn lagging_subscriber_starts_over_from_a_fresh_state() {
267	let harness = start(EngineConfig::default(), 4).await;
268	let transaction_id = TransactionId::new();
269	let mut updates = harness.mirror.subscribe();
270	assert!(matches!(
271	updates.next().await.unwrap().change,
272	StateChange::State(_)
273	));
274	for uploaded_bytes in 1..=20 {
275	harness
276	.inject
277	.send(Event::MediaUploadProgress {
278	room_id: harness.room_id.clone(),
279	transaction_id: transaction_id.clone(),
280	uploaded_bytes,
281	total_bytes: 20,
282	complete: uploaded_bytes == 20,
283	})
284	.unwrap();
285	}
286	tokio::time::timeout(Duration::from_secs(5), async {
287	while !harness
288	.mirror
289	.snapshot()
290	.uploads
291	.get(&transaction_id)
292	.is_some_and(\|upload\| upload.complete)
293	{
294	tokio::task::yield_now().await;
295	}
296	})
297	.await
298	.expect("the mirror never caught up");
299
300	let update = updates.next().await.unwrap();
301	let StateChange::State(state) = &update.change else {
302	panic!("a lagged subscription carried on with {update:?}");
303	};
304	assert!(state.uploads[&transaction_id].complete);
305	}
306
307	#[tokio::test]
308	async fn resync_does_not_deadlock_on_full_queues() {
309	let config = EngineConfig::default()
310	.with_command_capacity(1)
311	.with_event_capacity(1);
312	let harness = start(config, 1024).await;
313	let timeline_id = TimelineId::from(harness.room_id.clone());
314	let mut updates = harness.mirror.subscribe();
315	harness
316	.handle
317	.open_room(harness.room_id.clone())
318	.await
319	.unwrap();
320	next_matching(&mut updates, \|change\| {
321	timeline_change(change, &timeline_id, \|change\| {
322	matches!(change, TimelineChange::Reset { .. })
323	})
324	})
325	.await;
326
327	let flood = tokio::spawn({
328	let handle = harness.handle.clone();
329	let timeline_id = timeline_id.clone();
330	async move {
331	loop {
332	let command = Command::Timeline(TimelineCommand::Paginate {
333	timeline_id: timeline_id.clone(),
334	limit: 1,
335	});
336	if handle.send(command).await.is_err() {
337	return;
338	}
339	}
340	}
341	});
342	for _ in 0..10 {
343	harness
344	.inject
345	.send(Event::TimelineChanged {
346	timeline_id: timeline_id.clone(),
347	diffs: vec![TimelineDiff::Remove { index: 1000 }],
348	})
349	.unwrap();
350	next_matching(&mut updates, \|change\| {
351	timeline_change(change, &timeline_id, \|change\| {
352	matches!(change, TimelineChange::Diverged)
353	})
354	})
355	.await;
356	next_matching(&mut updates, \|change\| {
357	timeline_change(change, &timeline_id, \|change\| {
358	matches!(change, TimelineChange::Reset { .. })
359	})
360	})
361	.await;
362	}
363	flood.abort();
364	}
365
366	#[tokio::test]
367	async fn closing_a_timeline_evicts_it() {
368	let harness = start(EngineConfig::default(), 256).await;
369	let timeline_id = TimelineId::from(harness.room_id.clone());
370	let mut updates = harness.mirror.subscribe();
371	harness
372	.handle
373	.open_room(harness.room_id.clone())
374	.await
375	.unwrap();
376	harness
377	.handle
378	.close_room(harness.room_id.clone())
379	.await
380	.unwrap();
381	next_matching(&mut updates, \|change\| {
382	timeline_change(change, &timeline_id, \|change\| {
383	matches!(change, TimelineChange::Closed)
384	})
385	})
386	.await;
387	assert!(
388	!harness
389	.mirror
390	.snapshot()
391	.timelines
392	.contains_key(&timeline_id)
393	);
394	}
395
396	#[tokio::test]
397	async fn an_upload_ends_on_finished() {
398	let harness = start(EngineConfig::default(), 256).await;
399	let transaction_id = TransactionId::new();
400	let mut updates = harness.mirror.subscribe();
401	harness
402	.inject
403	.send(Event::MediaUploadProgress {
404	room_id: harness.room_id.clone(),
405	transaction_id: transaction_id.clone(),
406	uploaded_bytes: 5,
407	total_bytes: 10,
408	complete: false,
409	})
410	.unwrap();
411	next_matching(&mut updates, \|change\| {
412	matches!(change, StateChange::Upload { .. })
413	})
414	.await;
415	assert!(
416	harness
417	.mirror
418	.snapshot()
419	.uploads
420	.contains_key(&transaction_id)
421	);
422	harness
423	.inject
424	.send(Event::MediaUploadFinished {
425	room_id: harness.room_id.clone(),
426	transaction_id: transaction_id.clone(),
427	outcome: UploadOutcome::Cancelled,
428	})
429	.unwrap();
430	next_matching(&mut updates, \|change\| {
431	matches!(change, StateChange::UploadFinished { .. })
432	})
433	.await;
434	assert!(harness.mirror.snapshot().uploads.is_empty());
435	}