macltok

public

Star0

Fork 0

Watch0

12 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `7c26fad34822`

tarball

atagen <boss@atagen.co>2026-09-27 01:42 UTC

5 files changed+215-99

engine: give every reported upload an end, and the device to log back in as

Progress went out straight from the room forwarders, so an upload the tracker lost to a lagged client-wide channel could get progress and never a MediaUploadFinished. Progress now passes through the tracker, which registers the id, drops ticks for uploads it has already ended, and is the only task emitting upload events, so nothing about an upload can reach the stream after its end. SessionAccessRejected carries the device on a soft logout, so a stream-only consumer needn't read it from the record.

Modified file: docs/design-decisions.md+7 additions@@ -152,6 +152,13 @@ missed when that channel overflowed at the wrong moment, since then there's no
 knowing which. a failure isn't an ending, a wedged upload can still be retried.
 uploads don't belong to a room's actor, so closing the room leaves them be
 
+the forwarders hand their progress to that watcher rather than the stream, so
+it's the only thing speaking about uploads. anything that ever showed progress
+or a failure is followed until it gets exactly one `MediaUploadFinished`, and
+nothing about it goes out after. it remembers what has ended for as long as it
+runs, since a forwarder can still be sitting on progress for an upload that's
+already sent and nothing says when that backlog is gone
+
 ## sessions and secrets
 
 `Session::login_password`, `begin_oauth` and `restore` share one client builder

Modified file: src/engine/mod.rs+8 additions−2 deletions@@ -219,6 +219,7 @@ struct Queues {
     notifications: mpsc::Receiver<Job<NotificationCommand>>,
     encryption: mpsc::Receiver<Job<EncryptionCommand>>,
     session_changes: broadcast::Receiver<SessionChange>,
+    upload_progress: mpsc::Receiver<watchers::UploadProgress>,
 }
 
 /// requests other devices send us, handed from the sdk's event handler to the encryption worker
@@ -244,6 +245,7 @@ pub struct Engine {
     queues: Option<Queues>,
     /// one progress forwarder per room that has uploaded, and whether it's subscribed yet
     uploads: HashMap<OwnedRoomId, watch::Receiver<bool>>,
+    upload_progress: mpsc::Sender<watchers::UploadProgress>,
     /// the client can outlive us, so the handler has to come off it when we stop
     verification_handler: Option<matrix_sdk::event_handler::EventHandlerHandle>,
     stop: mpsc::UnboundedSender<Termination>,
@@ -288,6 +290,7 @@ impl Engine {
         client.send_queue().enable_upload_progress(true);
         let (note_sender, notes) = mpsc::unbounded_channel();
         let (stop, stops) = mpsc::unbounded_channel();
+        let (upload_progress, upload_progress_queue) = mpsc::channel(config.event_capacity);
         let (mutations, mutation_queue) = mpsc::channel(config.command_capacity);
         let (notifications, notification_queue) = mpsc::channel(config.command_capacity);
         let (encryption, encryption_queue) = mpsc::channel(config.command_capacity);
@@ -313,7 +316,9 @@ impl Engine {
                 notifications: notification_queue,
                 encryption: encryption_queue,
                 session_changes,
+                upload_progress: upload_progress_queue,
             }),
+            upload_progress,
             uploads: HashMap::new(),
             verification_handler: None,
             stop,
@@ -401,10 +406,11 @@ impl Engine {
         let events = || self.port.event_sender();
         let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 11] = [
             (
-                "upload watcher",
+                "upload tracker",
                 Box::pin(watchers::uploads(
                     client.clone(),
                     client.send_queue().subscribe(),
+                    queues.upload_progress,
                     events(),
                 )),
             ),
@@ -648,7 +654,7 @@ impl Engine {
         let (subscribed, ready) = watch::channel(false);
         self.tasks.spawn(
             TaskContext::Command("upload watcher"),
-            watchers::room_uploads(room.clone(), self.port.event_sender(), subscribed),
+            watchers::room_uploads(room.clone(), self.upload_progress.clone(), subscribed),
         );
         self.uploads
             .insert(room.room_id().to_owned(), ready.clone());

Modified file: src/engine/watchers.rs+190 additions−92 deletions@@ -42,15 +42,15 @@ pub(super) async fn session(
         let event = match changes.recv().await {
             Ok(SessionChange::TokensRefreshed) => Event::SessionTokensChanged,
             Ok(SessionChange::UnknownToken(data)) => {
+                let soft_logout = data.soft_logout.then_some(device_id);
                 let _ = events
                     .send(Event::SessionAccessRejected {
-                        soft_logout: data.soft_logout,
+                        soft_logout: soft_logout.clone(),
                     })
                     .await;
-                let _ = stop.send(if data.soft_logout {
-                    Termination::SoftLoggedOut { device_id }
-                } else {
-                    Termination::AccessRevoked
+                let _ = stop.send(match soft_logout {
+                    Some(device_id) => Termination::SoftLoggedOut { device_id },
+                    None => Termination::AccessRevoked,
                 });
                 return;
             }
@@ -146,7 +146,11 @@ pub(super) async fn send_queue_errors(
 }
 
 /// 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>) {
+pub(super) async fn room_uploads(
+    room: Room,
+    progress: mpsc::Sender<UploadProgress>,
+    ready: watch::Sender<bool>,
+) {
     let room_id = room.room_id().to_owned();
     let mut updates = match room.send_queue().subscribe().await {
         Ok((_, updates)) => updates,
@@ -166,66 +170,68 @@ pub(super) async fn room_uploads(room: Room, events: EventSender, ready: watch::
         let RoomSendQueueUpdate::MediaUpload {
             related_to,
             file,
-            progress,
+            progress: done,
             ..
         } = 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,
+        let tick = UploadProgress {
+            upload: (room_id.clone(), related_to),
+            uploaded_bytes: done.current as u64,
+            total_bytes: done.total as u64,
+            complete: file.is_some() && done.current == done.total,
         };
-        if events.send(event).await.is_err() {
+        if progress.send(tick).await.is_err() {
             return;
         }
     }
 }
 
-/// progress only goes out on each room's own channel, everything else reaches this
-/// one, which exists before any upload does
+pub(super) struct UploadProgress {
+    upload: Upload,
+    uploaded_bytes: u64,
+    total_bytes: u64,
+    complete: bool,
+}
+
+/// the one place upload events leave from, so an upload is never reported after it
+/// ended. progress comes in from the room forwarders, the rest from the client-wide
+/// channel, which exists before any upload does
 pub(super) async fn uploads(
     client: Client,
     mut updates: tokio::sync::broadcast::Receiver<SendQueueUpdate>,
+    mut progress: mpsc::Receiver<UploadProgress>,
     events: EventSender,
 ) {
     let (_, mut rooms) = client.rooms_stream();
-    let mut uploads = attachment_echoes(&client).await;
+    let mut tracker = UploadTracker {
+        live: attachment_echoes(&client).await,
+        ended: HashSet::new(),
+    };
     loop {
-        let finished = tokio::select! {
+        let reports = tokio::select! {
             update = updates.recv() => match update {
-                Ok(update) => match upload_update(&mut uploads, update) {
-                    Some(event) => vec![event],
-                    None => continue,
-                },
+                Ok(update) => tracker.update(update).into_iter().collect(),
                 Err(RecvError::Lagged(skipped)) => {
                     tracing::warn!(skipped, "upload watcher lagged");
                     let current = attachment_echoes(&client).await;
-                    let missed = uploads.difference(&current).cloned().collect::<Vec<_>>();
-                    uploads = current;
-                    finished_uploads(missed, UploadOutcome::Missed)
+                    tracker.resnapshot(current)
                 }
                 Err(RecvError::Closed) => return,
             },
+            Some(tick) = progress.recv() => tracker.progress(tick).into_iter().collect(),
             update = rooms.next() => {
                 if update.is_none() {
                     return;
                 }
-                let forgotten = uploads
-                    .iter()
-                    .filter(|(room_id, _)| client.get_room(room_id).is_none())
-                    .cloned()
-                    .collect::<Vec<_>>();
-                for upload in &forgotten {
-                    uploads.remove(upload);
-                }
-                finished_uploads(forgotten, UploadOutcome::RoomForgotten)
+                tracker.finish_where(
+                    |(room_id, _)| client.get_room(room_id).is_none(),
+                    &UploadOutcome::RoomForgotten,
+                )
             }
         };
-        for event in finished {
+        for event in reports {
             if events.send(event).await.is_err() {
                 return;
             }
@@ -235,75 +241,119 @@ pub(super) async fn uploads(
 
 type Upload = (OwnedRoomId, OwnedTransactionId);
 
-async fn attachment_echoes(client: &Client) -> HashSet<Upload> {
-    let Ok(echoes) = client.send_queue().local_echoes().await else {
-        return HashSet::new();
-    };
-    echoes
-        .into_iter()
-        .flat_map(|(room_id, echoes)| {
-            echoes
-                .into_iter()
-                .filter(|echo| is_attachment_echo(&echo.content))
-                .map(move |echo| (room_id.clone(), echo.transaction_id))
-        })
-        .collect()
+/// everything reported gets exactly one `MediaUploadFinished`. `ended` is kept for good,
+/// since a room forwarder can still be holding progress for an upload the client-wide
+/// channel has already seen sent
+struct UploadTracker {
+    live: HashSet<Upload>,
+    ended: HashSet<Upload>,
 }
 
-fn upload_update(uploads: &mut HashSet<Upload>, update: SendQueueUpdate) -> Option<Event> {
-    let room_id = update.room_id;
-    let (transaction_id, outcome) = match update.update {
-        RoomSendQueueUpdate::NewLocalEvent(echo) => {
-            if is_attachment_echo(&echo.content) {
-                uploads.insert((room_id, echo.transaction_id));
-            }
-            return None;
-        }
-        RoomSendQueueUpdate::SendError {
-            transaction_id,
-            error,
-            is_recoverable,
-        } => {
-            let upload = (room_id, transaction_id);
-            if !uploads.contains(&upload) {
+impl UploadTracker {
+    fn update(&mut self, update: SendQueueUpdate) -> Option<Event> {
+        let room_id = update.room_id;
+        let (transaction_id, outcome) = match update.update {
+            RoomSendQueueUpdate::NewLocalEvent(echo) => {
+                let upload = (room_id, echo.transaction_id);
+                if is_attachment_echo(&echo.content) && !self.ended.contains(&upload) {
+                    self.live.insert(upload);
+                }
                 return None;
             }
-            let (room_id, transaction_id) = upload;
-            return Some(Event::MediaUploadFailed {
-                room_id,
+            RoomSendQueueUpdate::SendError {
                 transaction_id,
-                kind: classify_sdk_error(&error),
-                will_retry: is_recoverable,
-            });
+                error,
+                is_recoverable,
+            } => {
+                let upload = (room_id, transaction_id);
+                if !self.live.contains(&upload) {
+                    return None;
+                }
+                let (room_id, transaction_id) = upload;
+                return Some(Event::MediaUploadFailed {
+                    room_id,
+                    transaction_id,
+                    kind: classify_sdk_error(&error),
+                    will_retry: is_recoverable,
+                });
+            }
+            RoomSendQueueUpdate::SentEvent {
+                transaction_id,
+                event_id,
+            } => (transaction_id, UploadOutcome::Sent { event_id }),
+            RoomSendQueueUpdate::CancelledLocalEvent { transaction_id } => {
+                (transaction_id, UploadOutcome::Cancelled)
+            }
+            _ => return None,
+        };
+        self.finish((room_id, transaction_id), outcome)
+    }
+
+    fn progress(&mut self, tick: UploadProgress) -> Option<Event> {
+        if self.ended.contains(&tick.upload) {
+            return None;
         }
-        RoomSendQueueUpdate::SentEvent {
+        self.live.insert(tick.upload.clone());
+        let (room_id, transaction_id) = tick.upload;
+        Some(Event::MediaUploadProgress {
+            room_id,
             transaction_id,
-            event_id,
-        } => (transaction_id, UploadOutcome::Sent { event_id }),
-        RoomSendQueueUpdate::CancelledLocalEvent { transaction_id } => {
-            (transaction_id, UploadOutcome::Cancelled)
+            uploaded_bytes: tick.uploaded_bytes,
+            total_bytes: tick.total_bytes,
+            complete: tick.complete,
+        })
+    }
+
+    fn resnapshot(&mut self, current: HashSet<Upload>) -> Vec<Event> {
+        let reports = self.finish_where(|upload| !current.contains(upload), &UploadOutcome::Missed);
+        let ended = &self.ended;
+        self.live
+            .extend(current.into_iter().filter(|upload| !ended.contains(upload)));
+        reports
+    }
+
+    fn finish_where(
+        &mut self,
+        ends: impl Fn(&Upload) -> bool,
+        outcome: &UploadOutcome,
+    ) -> Vec<Event> {
+        let ending = self
+            .live
+            .iter()
+            .filter(|upload| ends(upload))
+            .cloned()
+            .collect::<Vec<_>>();
+        ending
+            .into_iter()
+            .filter_map(|upload| self.finish(upload, outcome.clone()))
+            .collect()
+    }
+
+    fn finish(&mut self, upload: Upload, outcome: UploadOutcome) -> Option<Event> {
+        if !self.live.remove(&upload) {
+            return None;
         }
-        _ => return None,
-    };
-    let upload = (room_id, transaction_id);
-    if !uploads.remove(&upload) {
-        return None;
+        self.ended.insert(upload.clone());
+        let (room_id, transaction_id) = upload;
+        Some(Event::MediaUploadFinished {
+            room_id,
+            transaction_id,
+            outcome,
+        })
     }
-    let (room_id, transaction_id) = upload;
-    Some(Event::MediaUploadFinished {
-        room_id,
-        transaction_id,
-        outcome,
-    })
 }
 
-fn finished_uploads(uploads: Vec<Upload>, outcome: UploadOutcome) -> Vec<Event> {
-    uploads
+async fn attachment_echoes(client: &Client) -> HashSet<Upload> {
+    let Ok(echoes) = client.send_queue().local_echoes().await else {
+        return HashSet::new();
+    };
+    echoes
         .into_iter()
-        .map(|(room_id, transaction_id)| Event::MediaUploadFinished {
-            room_id,
-            transaction_id,
-            outcome: outcome.clone(),
+        .flat_map(|(room_id, echoes)| {
+            echoes
+                .into_iter()
+                .filter(|echo| is_attachment_echo(&echo.content))
+                .map(move |echo| (room_id.clone(), echo.transaction_id))
         })
         .collect()
 }
@@ -389,3 +439,51 @@ pub(super) async fn notification_settings(client: Client, events: EventSender) {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use matrix_sdk::send_queue::{RoomSendQueueUpdate, SendQueueUpdate};
+    use ruma::{OwnedTransactionId, event_id, owned_room_id};
+
+    use super::{HashSet, UploadProgress, UploadTracker};
+    use crate::{Event, UploadOutcome};
+
+    #[test]
+    fn an_upload_only_progress_saw_still_ends_once() {
+        let upload = (
+            owned_room_id!("!room:localhost"),
+            OwnedTransactionId::from("upload"),
+        );
+        let tick = || UploadProgress {
+            upload: upload.clone(),
+            uploaded_bytes: 1,
+            total_bytes: 2,
+            complete: false,
+        };
+        let mut tracker = UploadTracker {
+            live: HashSet::new(),
+            ended: HashSet::new(),
+        };
+        assert!(tracker.progress(tick()).is_some());
+
+        let missed = tracker.resnapshot(HashSet::new());
+        assert!(matches!(
+            missed.as_slice(),
+            [Event::MediaUploadFinished {
+                outcome: UploadOutcome::Missed,
+                ..
+            }]
+        ));
+
+        assert!(tracker.progress(tick()).is_none());
+        let sent = SendQueueUpdate {
+            room_id: upload.0.clone(),
+            update: RoomSendQueueUpdate::SentEvent {
+                transaction_id: upload.1.clone(),
+                event_id: event_id!("$sent:localhost").to_owned(),
+            },
+        };
+        assert!(tracker.update(sent).is_none());
+        assert!(tracker.resnapshot(HashSet::from([upload])).is_empty());
+    }
+}

Modified file: src/model.rs+4 additions−4 deletions@@ -637,11 +637,11 @@ pub enum Event {
     /// Refresh tokens changed. Persist a fresh `ClientHandle::session_record()`
     /// atomically; a record captured before this event may no longer restore.
     SessionTokensChanged,
-    /// the server stopped honouring our token and the engine stops. on a soft
-    /// logout keep the store and log in again with the same device, otherwise
-    /// the record and the store are both dead
+    /// the server stopped honouring our token and the engine stops. a soft logout
+    /// names the device to log in again as, keeping the store. otherwise the record
+    /// and the store are both dead
     SessionAccessRejected {
-        soft_logout: bool,
+        soft_logout: Option<OwnedDeviceId>,
     },
     NotificationSettingsChanged,
     VerificationTargetChanged(VerificationTarget),

Modified file: tests/state.rs+6 additions−1 deletions@@ -90,7 +90,12 @@ async fn soft_logout_says_so_and_stops_the_engine() {
     .unwrap();
     let running = tokio::spawn(engine.run());
     support::receive_until(&mut events, |event| {
-        matches!(event, Event::SessionAccessRejected { soft_logout: true })
+        matches!(
+            event,
+            Event::SessionAccessRejected {
+                soft_logout: Some(_)
+            }
+        )
     })
     .await;
     support::receive_until(&mut events, |event| matches!(event, Event::Stopped)).await;

macltok

Commit 7c26fad34822

Commit `7c26fad34822`