macltok

public

Star0

Watch0

8 commits 3 branches 0 tags 0 releases3.0 MiB

use std::{sync::Arc, time::Duration};

use futures_util::StreamExt;
use matrix_sdk::{
    Client, Room, SessionChange,
    ruma::events::{AnyMessageLikeEventContent, room::message::MessageType},
    send_queue::{LocalEchoContent, RoomSendQueueUpdate},
};
use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};
use ruma::{OwnedDeviceId, time::Instant};
use tokio::sync::{broadcast::error::RecvError, mpsc, watch};

use super::executor::TaskSet;
use crate::{
    ConnectionState, Event, Termination, encryption::encryption_state, error::classify_sdk_error,
    transport::EventSender,
};

const SYNC_RESTART_MIN: Duration = Duration::from_secs(1);
const SYNC_RESTART_MAX: Duration = Duration::from_secs(60);
const SEND_QUEUE_RETRY: Duration = Duration::from_secs(5);
const SYNC_STABLE: Duration = Duration::from_secs(30);

pub(super) async fn room_subscriptions(
    sync: Arc<SyncService>,
    mut rooms: watch::Receiver<Vec<ruma::OwnedRoomId>>,
) {
    while rooms.changed().await.is_ok() {
        let room_ids = rooms.borrow_and_update().clone();
        let room_ids = room_ids.iter().map(AsRef::as_ref).collect::<Vec<_>>();
        sync.room_list_service().subscribe_to_rooms(&room_ids).await;
    }
}

pub(super) async fn session(
    mut changes: tokio::sync::broadcast::Receiver<SessionChange>,
    events: EventSender,
    device_id: OwnedDeviceId,
    stop: mpsc::UnboundedSender<Termination>,
) {
    loop {
        let event = match changes.recv().await {
            Ok(SessionChange::TokensRefreshed) => Event::SessionTokensChanged,
            Ok(SessionChange::UnknownToken(data)) => {
                let _ = events
                    .send(Event::SessionAccessRejected {
                        soft_logout: data.soft_logout,
                    })
                    .await;
                let _ = stop.send(if data.soft_logout {
                    Termination::SoftLoggedOut { device_id }
                } else {
                    Termination::AccessRevoked
                });
                return;
            }
            // a missed refresh costs one needless save, a missed rejection comes round again
            Err(RecvError::Lagged(_)) => Event::SessionTokensChanged,
            Err(RecvError::Closed) => return,
        };
        if events.send(event).await.is_err() {
            return;
        }
    }
}

/// `states` must be taken before sync starts, a late subscriber never hears the transition it missed
pub(super) async fn connection(
    mut states: impl futures_util::Stream<Item = SyncState> + Unpin,
    sync: Arc<SyncService>,
    client: Client,
    events: EventSender,
    connection: watch::Sender<ConnectionState>,
) {
    let mut backoff = SYNC_RESTART_MIN;
    let mut online_since: Option<Instant> = None;
    while let Some(state) = states.next().await {
        // only a run that lasted counts as recovered, or a server that answers /versions
        // but refuses sync would have us bounce between offline and online forever
        let stable = online_since
            .take()
            .map(|since| since.elapsed() >= SYNC_STABLE);
        if stable == Some(true) {
            backoff = SYNC_RESTART_MIN;
        }
        let (state, restart) = match state {
            // only ever us stopping it for a restart, nothing a client needs to hear
            SyncState::Idle => continue,
            SyncState::Running => {
                online_since = Some(Instant::now());
                // the sdk parks a room queue on any send error and never unparks it
                client.send_queue().set_enabled(true).await;
                (ConnectionState::Online, false)
            }
            SyncState::Offline => (ConnectionState::Offline, stable == Some(false)),
            SyncState::Terminated => (ConnectionState::Offline, true),
            SyncState::Error(error) => {
                tracing::warn!(%error, "sync service reported an error");
                (ConnectionState::Offline, true)
            }
        };
        if *connection.borrow() != state {
            connection.send_replace(state);
            if events.send(Event::ConnectionChanged(state)).await.is_err() {
                return;
            }
        }
        if restart {
            sync.stop().await;
            matrix_sdk::sleep::sleep(backoff).await;
            backoff = (backoff * 2).min(SYNC_RESTART_MAX);
            sync.start().await;
        }
    }
}

/// a recoverable failure while we're online is a hiccup, so give the room another go
pub(super) async fn send_queue_errors(
    client: Client,
    mut errors: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueRoomError>,
    online: watch::Receiver<ConnectionState>,
) {
    let mut retries = TaskSet::new();
    loop {
        let error = tokio::select! {
            error = errors.recv() => error,
            Some(_) = retries.join_next(), if !retries.is_empty() => continue,
        };
        match error {
            Ok(error) if error.is_recoverable => {
                let client = client.clone();
                let online = online.clone();
                retries.spawn(async move {
                    matrix_sdk::sleep::sleep(SEND_QUEUE_RETRY).await;
                    if *online.borrow() == ConnectionState::Online
                        && let Some(room) = client.get_room(&error.room_id)
                    {
                        room.send_queue().set_enabled(true);
                    }
                });
            }
            Ok(_) | Err(RecvError::Lagged(_)) => {}
            Err(RecvError::Closed) => return,
        }
    }
}

/// upload progress only goes out on the room's own channel, never the client-wide one
pub(super) async fn room_uploads(room: Room, events: EventSender, ready: watch::Sender<bool>) {
    let room_id = room.room_id().to_owned();
    let mut updates = match room.send_queue().subscribe().await {
        Ok((_, updates)) => updates,
        Err(error) => {
            tracing::warn!(%room_id, %error, "reading the media send queue failed");
            return;
        }
    };
    ready.send_replace(true);
    loop {
        let update = match updates.recv().await {
            Ok(update) => update,
            // a missed tick is fine, the next one says where it's got to
            Err(RecvError::Lagged(_)) => continue,
            Err(RecvError::Closed) => return,
        };
        let RoomSendQueueUpdate::MediaUpload {
            related_to,
            file,
            progress,
            ..
        } = update
        else {
            continue;
        };
        let event = Event::MediaUploadProgress {
            room_id: room_id.clone(),
            transaction_id: related_to,
            uploaded_bytes: progress.current as u64,
            total_bytes: progress.total as u64,
            complete: file.is_some() && progress.current == progress.total,
        };
        if events.send(event).await.is_err() {
            return;
        }
    }
}

/// failures do reach the client-wide channel, and this one exists before any upload does
pub(super) async fn upload_failures(
    client: Client,
    mut updates: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueUpdate>,
    events: EventSender,
) {
    loop {
        let update = match updates.recv().await {
            Ok(update) => update,
            Err(RecvError::Lagged(skipped)) => {
                tracing::warn!(skipped, "upload failure watcher lagged");
                continue;
            }
            Err(RecvError::Closed) => return,
        };
        let RoomSendQueueUpdate::SendError {
            transaction_id,
            error,
            is_recoverable,
        } = update.update
        else {
            continue;
        };
        let Some(room) = client.get_room(&update.room_id) else {
            continue;
        };
        let is_attachment = room
            .send_queue()
            .subscribe()
            .await
            .is_ok_and(|(echoes, _)| {
                echoes.iter().any(|echo| {
                    echo.transaction_id == transaction_id && is_attachment_echo(&echo.content)
                })
            });
        if !is_attachment {
            continue;
        }
        let event = Event::MediaUploadFailed {
            room_id: update.room_id,
            transaction_id,
            kind: classify_sdk_error(&error),
            recoverable: is_recoverable,
        };
        if events.send(event).await.is_err() {
            return;
        }
    }
}

fn is_attachment_echo(content: &LocalEchoContent) -> bool {
    let LocalEchoContent::Event {
        serialized_event, ..
    } = content
    else {
        return false;
    };
    matches!(
        serialized_event.deserialize(),
        Ok(AnyMessageLikeEventContent::RoomMessage(content))
            if matches!(
                content.msgtype,
                MessageType::Audio(_) | MessageType::File(_) | MessageType::Image(_) | MessageType::Video(_)
            )
    )
}

pub(super) async fn encryption(client: Client, events: EventSender) {
    let Some(device_id) = client.device_id().map(ToOwned::to_owned) else {
        tracing::warn!("encryption watcher stopped because the client has no device id");
        return;
    };
    let recovery = client.encryption().recovery();
    let backups = client.encryption().backups();
    let mut recovery_stream = recovery.state_stream();
    let mut backup_stream = backups.state_stream();
    let mut verification_stream = client.encryption().verification_state();
    let mut recovery_state = recovery.state();
    let mut backup_state = backups.state();
    let mut verification_state = verification_stream.get();
    loop {
        let state = encryption_state(
            device_id.clone(),
            recovery_state,
            verification_state,
            backup_state,
        );
        if events.send(Event::EncryptionChanged(state)).await.is_err() {
            return;
        }
        tokio::select! {
            Some(update) = recovery_stream.next() => recovery_state = update,
            Some(update) = verification_stream.next() => verification_state = update,
            Some(update) = backup_stream.next() => {
                backup_state = update.unwrap_or_else(|_| backups.state());
            }
            else => return,
        }
    }
}

pub(super) async fn rooms(client: Client, events: EventSender) {
    let (_, mut rooms) = client.rooms_stream();
    let mut info = client.room_info_notable_update_receiver();
    loop {
        if events.send(Event::RoomsChanged).await.is_err() {
            return;
        }
        tokio::select! {
            update = rooms.next() => if update.is_none() { return },
            update = info.recv() => if matches!(update, Err(RecvError::Closed)) { return },
        }
        // a sync touching fifty rooms is one invalidation, not fifty
        while let Ok(_) | Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) =
            info.try_recv()
        {}
    }
}

pub(super) async fn notification_settings(client: Client, events: EventSender) {
    let settings = client.notification_settings().await;
    let mut changes = settings.subscribe_to_changes();
    while let Ok(()) | Err(RecvError::Lagged(_)) = changes.recv().await {
        if events
            .send(Event::NotificationSettingsChanged)
            .await
            .is_err()
        {
            return;
        }
    }
}

1	use std::{sync::Arc, time::Duration};
2
3	use futures_util::StreamExt;
4	use matrix_sdk::{
5	Client, Room, SessionChange,
6	ruma::events::{AnyMessageLikeEventContent, room::message::MessageType},
7	send_queue::{LocalEchoContent, RoomSendQueueUpdate},
8	};
9	use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};
10	use ruma::{OwnedDeviceId, time::Instant};
11	use tokio::sync::{broadcast::error::RecvError, mpsc, watch};
12
13	use super::executor::TaskSet;
14	use crate::{
15	ConnectionState, Event, Termination, encryption::encryption_state, error::classify_sdk_error,
16	transport::EventSender,
17	};
18
19	const SYNC_RESTART_MIN: Duration = Duration::from_secs(1);
20	const SYNC_RESTART_MAX: Duration = Duration::from_secs(60);
21	const SEND_QUEUE_RETRY: Duration = Duration::from_secs(5);
22	const SYNC_STABLE: Duration = Duration::from_secs(30);
23
24	pub(super) async fn room_subscriptions(
25	sync: Arc<SyncService>,
26	mut rooms: watch::Receiver<Vec<ruma::OwnedRoomId>>,
27	) {
28	while rooms.changed().await.is_ok() {
29	let room_ids = rooms.borrow_and_update().clone();
30	let room_ids = room_ids.iter().map(AsRef::as_ref).collect::<Vec<_>>();
31	sync.room_list_service().subscribe_to_rooms(&room_ids).await;
32	}
33	}
34
35	pub(super) async fn session(
36	mut changes: tokio::sync::broadcast::Receiver<SessionChange>,
37	events: EventSender,
38	device_id: OwnedDeviceId,
39	stop: mpsc::UnboundedSender<Termination>,
40	) {
41	loop {
42	let event = match changes.recv().await {
43	Ok(SessionChange::TokensRefreshed) => Event::SessionTokensChanged,
44	Ok(SessionChange::UnknownToken(data)) => {
45	let _ = events
46	.send(Event::SessionAccessRejected {
47	soft_logout: data.soft_logout,
48	})
49	.await;
50	let _ = stop.send(if data.soft_logout {
51	Termination::SoftLoggedOut { device_id }
52	} else {
53	Termination::AccessRevoked
54	});
55	return;
56	}
57	// a missed refresh costs one needless save, a missed rejection comes round again
58	Err(RecvError::Lagged(_)) => Event::SessionTokensChanged,
59	Err(RecvError::Closed) => return,
60	};
61	if events.send(event).await.is_err() {
62	return;
63	}
64	}
65	}
66
67	/// `states` must be taken before sync starts, a late subscriber never hears the transition it missed
68	pub(super) async fn connection(
69	mut states: impl futures_util::Stream<Item = SyncState> + Unpin,
70	sync: Arc<SyncService>,
71	client: Client,
72	events: EventSender,
73	connection: watch::Sender<ConnectionState>,
74	) {
75	let mut backoff = SYNC_RESTART_MIN;
76	let mut online_since: Option<Instant> = None;
77	while let Some(state) = states.next().await {
78	// only a run that lasted counts as recovered, or a server that answers /versions
79	// but refuses sync would have us bounce between offline and online forever
80	let stable = online_since
81	.take()
82	.map(\|since\| since.elapsed() >= SYNC_STABLE);
83	if stable == Some(true) {
84	backoff = SYNC_RESTART_MIN;
85	}
86	let (state, restart) = match state {
87	// only ever us stopping it for a restart, nothing a client needs to hear
88	SyncState::Idle => continue,
89	SyncState::Running => {
90	online_since = Some(Instant::now());
91	// the sdk parks a room queue on any send error and never unparks it
92	client.send_queue().set_enabled(true).await;
93	(ConnectionState::Online, false)
94	}
95	SyncState::Offline => (ConnectionState::Offline, stable == Some(false)),
96	SyncState::Terminated => (ConnectionState::Offline, true),
97	SyncState::Error(error) => {
98	tracing::warn!(%error, "sync service reported an error");
99	(ConnectionState::Offline, true)
100	}
101	};
102	if *connection.borrow() != state {
103	connection.send_replace(state);
104	if events.send(Event::ConnectionChanged(state)).await.is_err() {
105	return;
106	}
107	}
108	if restart {
109	sync.stop().await;
110	matrix_sdk::sleep::sleep(backoff).await;
111	backoff = (backoff * 2).min(SYNC_RESTART_MAX);
112	sync.start().await;
113	}
114	}
115	}
116
117	/// a recoverable failure while we're online is a hiccup, so give the room another go
118	pub(super) async fn send_queue_errors(
119	client: Client,
120	mut errors: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueRoomError>,
121	online: watch::Receiver<ConnectionState>,
122	) {
123	let mut retries = TaskSet::new();
124	loop {
125	let error = tokio::select! {
126	error = errors.recv() => error,
127	Some(_) = retries.join_next(), if !retries.is_empty() => continue,
128	};
129	match error {
130	Ok(error) if error.is_recoverable => {
131	let client = client.clone();
132	let online = online.clone();
133	retries.spawn(async move {
134	matrix_sdk::sleep::sleep(SEND_QUEUE_RETRY).await;
135	if *online.borrow() == ConnectionState::Online
136	&& let Some(room) = client.get_room(&error.room_id)
137	{
138	room.send_queue().set_enabled(true);
139	}
140	});
141	}
142	Ok(_) \| Err(RecvError::Lagged(_)) => {}
143	Err(RecvError::Closed) => return,
144	}
145	}
146	}
147
148	/// upload progress only goes out on the room's own channel, never the client-wide one
149	pub(super) async fn room_uploads(room: Room, events: EventSender, ready: watch::Sender<bool>) {
150	let room_id = room.room_id().to_owned();
151	let mut updates = match room.send_queue().subscribe().await {
152	Ok((_, updates)) => updates,
153	Err(error) => {
154	tracing::warn!(%room_id, %error, "reading the media send queue failed");
155	return;
156	}
157	};
158	ready.send_replace(true);
159	loop {
160	let update = match updates.recv().await {
161	Ok(update) => update,
162	// a missed tick is fine, the next one says where it's got to
163	Err(RecvError::Lagged(_)) => continue,
164	Err(RecvError::Closed) => return,
165	};
166	let RoomSendQueueUpdate::MediaUpload {
167	related_to,
168	file,
169	progress,
170	..
171	} = update
172	else {
173	continue;
174	};
175	let event = Event::MediaUploadProgress {
176	room_id: room_id.clone(),
177	transaction_id: related_to,
178	uploaded_bytes: progress.current as u64,
179	total_bytes: progress.total as u64,
180	complete: file.is_some() && progress.current == progress.total,
181	};
182	if events.send(event).await.is_err() {
183	return;
184	}
185	}
186	}
187
188	/// failures do reach the client-wide channel, and this one exists before any upload does
189	pub(super) async fn upload_failures(
190	client: Client,
191	mut updates: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueUpdate>,
192	events: EventSender,
193	) {
194	loop {
195	let update = match updates.recv().await {
196	Ok(update) => update,
197	Err(RecvError::Lagged(skipped)) => {
198	tracing::warn!(skipped, "upload failure watcher lagged");
199	continue;
200	}
201	Err(RecvError::Closed) => return,
202	};
203	let RoomSendQueueUpdate::SendError {
204	transaction_id,
205	error,
206	is_recoverable,
207	} = update.update
208	else {
209	continue;
210	};
211	let Some(room) = client.get_room(&update.room_id) else {
212	continue;
213	};
214	let is_attachment = room
215	.send_queue()
216	.subscribe()
217	.await
218	.is_ok_and(\|(echoes, _)\| {
219	echoes.iter().any(\|echo\| {
220	echo.transaction_id == transaction_id && is_attachment_echo(&echo.content)
221	})
222	});
223	if !is_attachment {
224	continue;
225	}
226	let event = Event::MediaUploadFailed {
227	room_id: update.room_id,
228	transaction_id,
229	kind: classify_sdk_error(&error),
230	recoverable: is_recoverable,
231	};
232	if events.send(event).await.is_err() {
233	return;
234	}
235	}
236	}
237
238	fn is_attachment_echo(content: &LocalEchoContent) -> bool {
239	let LocalEchoContent::Event {
240	serialized_event, ..
241	} = content
242	else {
243	return false;
244	};
245	matches!(
246	serialized_event.deserialize(),
247	Ok(AnyMessageLikeEventContent::RoomMessage(content))
248	if matches!(
249	content.msgtype,
250	MessageType::Audio(_) \| MessageType::File(_) \| MessageType::Image(_) \| MessageType::Video(_)
251	)
252	)
253	}
254
255	pub(super) async fn encryption(client: Client, events: EventSender) {
256	let Some(device_id) = client.device_id().map(ToOwned::to_owned) else {
257	tracing::warn!("encryption watcher stopped because the client has no device id");
258	return;
259	};
260	let recovery = client.encryption().recovery();
261	let backups = client.encryption().backups();
262	let mut recovery_stream = recovery.state_stream();
263	let mut backup_stream = backups.state_stream();
264	let mut verification_stream = client.encryption().verification_state();
265	let mut recovery_state = recovery.state();
266	let mut backup_state = backups.state();
267	let mut verification_state = verification_stream.get();
268	loop {
269	let state = encryption_state(
270	device_id.clone(),
271	recovery_state,
272	verification_state,
273	backup_state,
274	);
275	if events.send(Event::EncryptionChanged(state)).await.is_err() {
276	return;
277	}
278	tokio::select! {
279	Some(update) = recovery_stream.next() => recovery_state = update,
280	Some(update) = verification_stream.next() => verification_state = update,
281	Some(update) = backup_stream.next() => {
282	backup_state = update.unwrap_or_else(\|_\| backups.state());
283	}
284	else => return,
285	}
286	}
287	}
288
289	pub(super) async fn rooms(client: Client, events: EventSender) {
290	let (_, mut rooms) = client.rooms_stream();
291	let mut info = client.room_info_notable_update_receiver();
292	loop {
293	if events.send(Event::RoomsChanged).await.is_err() {
294	return;
295	}
296	tokio::select! {
297	update = rooms.next() => if update.is_none() { return },
298	update = info.recv() => if matches!(update, Err(RecvError::Closed)) { return },
299	}
300	// a sync touching fifty rooms is one invalidation, not fifty
301	while let Ok(_) \| Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) =
302	info.try_recv()
303	{}
304	}
305	}
306
307	pub(super) async fn notification_settings(client: Client, events: EventSender) {
308	let settings = client.notification_settings().await;
309	let mut changes = settings.subscribe_to_changes();
310	while let Ok(()) \| Err(RecvError::Lagged(_)) = changes.recv().await {
311	if events
312	.send(Event::NotificationSettingsChanged)
313	.await
314	.is_err()
315	{
316	return;
317	}
318	}
319	}