macltok

public

Star0

Watch0

3 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 tokio::{
    sync::{Notify, broadcast::error::RecvError, watch},
    task::JoinSet,
};

use crate::{
    ConnectionState, Event, 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);

pub(super) async fn session(
    mut changes: tokio::sync::broadcast::Receiver<SessionChange>,
    events: EventSender,
    stop: Arc<Notify>,
) {
    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;
                stop.notify_one();
                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;
    while let Some(state) = states.next().await {
        let (state, restart) = match state {
            SyncState::Idle => (ConnectionState::Idle, false),
            SyncState::Running => {
                backoff = SYNC_RESTART_MIN;
                // 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, 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 {
            tokio::time::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 = JoinSet::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 {
                    tokio::time::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,
            Err(RecvError::Lagged(skipped)) => {
                tracing::warn!(%room_id, skipped, "upload watcher lagged");
                continue;
            }
            Err(RecvError::Closed) => return,
        };
        let event = match update {
            RoomSendQueueUpdate::MediaUpload {
                related_to,
                file,
                progress,
                ..
            } => 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,
            },
            RoomSendQueueUpdate::SendError {
                transaction_id,
                error,
                is_recoverable,
            } => {
                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;
                }
                Event::MediaUploadFailed {
                    room_id: room_id.clone(),
                    transaction_id,
                    kind: classify_sdk_error(&error),
                    recoverable: is_recoverable,
                }
            }
            _ => continue,
        };
        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 tokio::{
11	sync::{Notify, broadcast::error::RecvError, watch},
12	task::JoinSet,
13	};
14
15	use crate::{
16	ConnectionState, Event, encryption::encryption_state, error::classify_sdk_error,
17	transport::EventSender,
18	};
19
20	const SYNC_RESTART_MIN: Duration = Duration::from_secs(1);
21	const SYNC_RESTART_MAX: Duration = Duration::from_secs(60);
22	const SEND_QUEUE_RETRY: Duration = Duration::from_secs(5);
23
24	pub(super) async fn session(
25	mut changes: tokio::sync::broadcast::Receiver<SessionChange>,
26	events: EventSender,
27	stop: Arc<Notify>,
28	) {
29	loop {
30	let event = match changes.recv().await {
31	Ok(SessionChange::TokensRefreshed) => Event::SessionTokensChanged,
32	Ok(SessionChange::UnknownToken(data)) => {
33	let _ = events
34	.send(Event::SessionAccessRejected {
35	soft_logout: data.soft_logout,
36	})
37	.await;
38	stop.notify_one();
39	return;
40	}
41	// a missed refresh costs one needless save, a missed rejection comes round again
42	Err(RecvError::Lagged(_)) => Event::SessionTokensChanged,
43	Err(RecvError::Closed) => return,
44	};
45	if events.send(event).await.is_err() {
46	return;
47	}
48	}
49	}
50
51	/// `states` must be taken before sync starts, a late subscriber never hears the transition it missed
52	pub(super) async fn connection(
53	mut states: impl futures_util::Stream<Item = SyncState> + Unpin,
54	sync: Arc<SyncService>,
55	client: Client,
56	events: EventSender,
57	connection: watch::Sender<ConnectionState>,
58	) {
59	let mut backoff = SYNC_RESTART_MIN;
60	while let Some(state) = states.next().await {
61	let (state, restart) = match state {
62	SyncState::Idle => (ConnectionState::Idle, false),
63	SyncState::Running => {
64	backoff = SYNC_RESTART_MIN;
65	// the sdk parks a room queue on any send error and never unparks it
66	client.send_queue().set_enabled(true).await;
67	(ConnectionState::Online, false)
68	}
69	SyncState::Offline => (ConnectionState::Offline, false),
70	SyncState::Terminated => (ConnectionState::Offline, true),
71	SyncState::Error(error) => {
72	tracing::warn!(%error, "sync service reported an error");
73	(ConnectionState::Offline, true)
74	}
75	};
76	if *connection.borrow() != state {
77	connection.send_replace(state);
78	if events.send(Event::ConnectionChanged(state)).await.is_err() {
79	return;
80	}
81	}
82	if restart {
83	tokio::time::sleep(backoff).await;
84	backoff = (backoff * 2).min(SYNC_RESTART_MAX);
85	sync.start().await;
86	}
87	}
88	}
89
90	/// a recoverable failure while we're online is a hiccup, so give the room another go
91	pub(super) async fn send_queue_errors(
92	client: Client,
93	mut errors: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueRoomError>,
94	online: watch::Receiver<ConnectionState>,
95	) {
96	let mut retries = JoinSet::new();
97	loop {
98	let error = tokio::select! {
99	error = errors.recv() => error,
100	Some(_) = retries.join_next(), if !retries.is_empty() => continue,
101	};
102	match error {
103	Ok(error) if error.is_recoverable => {
104	let client = client.clone();
105	let online = online.clone();
106	retries.spawn(async move {
107	tokio::time::sleep(SEND_QUEUE_RETRY).await;
108	if *online.borrow() == ConnectionState::Online
109	&& let Some(room) = client.get_room(&error.room_id)
110	{
111	room.send_queue().set_enabled(true);
112	}
113	});
114	}
115	Ok(_) \| Err(RecvError::Lagged(_)) => {}
116	Err(RecvError::Closed) => return,
117	}
118	}
119	}
120
121	/// upload progress only goes out on the room's own channel, never the client-wide one
122	pub(super) async fn room_uploads(room: Room, events: EventSender, ready: watch::Sender<bool>) {
123	let room_id = room.room_id().to_owned();
124	let mut updates = match room.send_queue().subscribe().await {
125	Ok((_, updates)) => updates,
126	Err(error) => {
127	tracing::warn!(%room_id, %error, "reading the media send queue failed");
128	return;
129	}
130	};
131	ready.send_replace(true);
132	loop {
133	let update = match updates.recv().await {
134	Ok(update) => update,
135	Err(RecvError::Lagged(skipped)) => {
136	tracing::warn!(%room_id, skipped, "upload watcher lagged");
137	continue;
138	}
139	Err(RecvError::Closed) => return,
140	};
141	let event = match update {
142	RoomSendQueueUpdate::MediaUpload {
143	related_to,
144	file,
145	progress,
146	..
147	} => Event::MediaUploadProgress {
148	room_id: room_id.clone(),
149	transaction_id: related_to,
150	uploaded_bytes: progress.current as u64,
151	total_bytes: progress.total as u64,
152	complete: file.is_some() && progress.current == progress.total,
153	},
154	RoomSendQueueUpdate::SendError {
155	transaction_id,
156	error,
157	is_recoverable,
158	} => {
159	let is_attachment = room
160	.send_queue()
161	.subscribe()
162	.await
163	.is_ok_and(\|(echoes, _)\| {
164	echoes.iter().any(\|echo\| {
165	echo.transaction_id == transaction_id
166	&& is_attachment_echo(&echo.content)
167	})
168	});
169	if !is_attachment {
170	continue;
171	}
172	Event::MediaUploadFailed {
173	room_id: room_id.clone(),
174	transaction_id,
175	kind: classify_sdk_error(&error),
176	recoverable: is_recoverable,
177	}
178	}
179	_ => continue,
180	};
181	if events.send(event).await.is_err() {
182	return;
183	}
184	}
185	}
186
187	fn is_attachment_echo(content: &LocalEchoContent) -> bool {
188	let LocalEchoContent::Event {
189	serialized_event, ..
190	} = content
191	else {
192	return false;
193	};
194	matches!(
195	serialized_event.deserialize(),
196	Ok(AnyMessageLikeEventContent::RoomMessage(content))
197	if matches!(
198	content.msgtype,
199	MessageType::Audio(_) \| MessageType::File(_) \| MessageType::Image(_) \| MessageType::Video(_)
200	)
201	)
202	}
203
204	pub(super) async fn encryption(client: Client, events: EventSender) {
205	let Some(device_id) = client.device_id().map(ToOwned::to_owned) else {
206	tracing::warn!("encryption watcher stopped because the client has no device id");
207	return;
208	};
209	let recovery = client.encryption().recovery();
210	let backups = client.encryption().backups();
211	let mut recovery_stream = recovery.state_stream();
212	let mut backup_stream = backups.state_stream();
213	let mut verification_stream = client.encryption().verification_state();
214	let mut recovery_state = recovery.state();
215	let mut backup_state = backups.state();
216	let mut verification_state = verification_stream.get();
217	loop {
218	let state = encryption_state(
219	device_id.clone(),
220	recovery_state,
221	verification_state,
222	backup_state,
223	);
224	if events.send(Event::EncryptionChanged(state)).await.is_err() {
225	return;
226	}
227	tokio::select! {
228	Some(update) = recovery_stream.next() => recovery_state = update,
229	Some(update) = verification_stream.next() => verification_state = update,
230	Some(update) = backup_stream.next() => {
231	backup_state = update.unwrap_or_else(\|_\| backups.state());
232	}
233	else => return,
234	}
235	}
236	}
237
238	pub(super) async fn rooms(client: Client, events: EventSender) {
239	let (_, mut rooms) = client.rooms_stream();
240	let mut info = client.room_info_notable_update_receiver();
241	loop {
242	if events.send(Event::RoomsChanged).await.is_err() {
243	return;
244	}
245	tokio::select! {
246	update = rooms.next() => if update.is_none() { return },
247	update = info.recv() => if matches!(update, Err(RecvError::Closed)) { return },
248	}
249	// a sync touching fifty rooms is one invalidation, not fifty
250	while let Ok(_) \| Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) =
251	info.try_recv()
252	{}
253	}
254	}
255
256	pub(super) async fn notification_settings(client: Client, events: EventSender) {
257	let settings = client.notification_settings().await;
258	let mut changes = settings.subscribe_to_changes();
259	while let Ok(()) \| Err(RecvError::Lagged(_)) = changes.recv().await {
260	if events
261	.send(Event::NotificationSettingsChanged)
262	.await
263	.is_err()
264	{
265	return;
266	}
267	}
268	}