macltok

public

Star0

Fork 0

Watch0

18 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `7cf348521204`

tarball

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

4 files changed+158-79

engine: never let an upload outlive its end

An end seen only on the client-wide channel for an upload the tracker hadn't met yet was dropped without a trace, so progress still queued in the room forwarder brought the upload back with nothing left to end it. Each forwarder now relays its room's sends and cancels behind its progress, which orders an upload's end after its ticks, and an end from anywhere else is remembered only until that relayed end arrives rather than for the whole run.

Modified file: docs/design-decisions.md+5 additions−3 deletions@@ -155,9 +155,11 @@ 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
+nothing about it goes out after. each forwarder relays its room's sends and
+cancels behind its progress, so an end never overtakes the ticks before it. an
+upload ended from anywhere else is remembered until its forwarder's own end
+turns up, since until then there may be progress still queued for it. the
+engine stopping ends nothing, `Stopped` covers it
 
 ## sessions and secrets
 

Modified file: src/engine/mod.rs+7 additions−7 deletions@@ -219,7 +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>,
+    upload_reports: mpsc::Receiver<watchers::RoomUpload>,
 }
 
 /// requests other devices send us, handed from the sdk's event handler to the encryption worker
@@ -245,7 +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>,
+    upload_reports: mpsc::Sender<watchers::RoomUpload>,
     /// 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>,
@@ -290,7 +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 (upload_reports, upload_report_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);
@@ -316,9 +316,9 @@ impl Engine {
                 notifications: notification_queue,
                 encryption: encryption_queue,
                 session_changes,
-                upload_progress: upload_progress_queue,
+                upload_reports: upload_report_queue,
             }),
-            upload_progress,
+            upload_reports,
             uploads: HashMap::new(),
             verification_handler: None,
             stop,
@@ -410,7 +410,7 @@ impl Engine {
                 Box::pin(watchers::uploads(
                     client.clone(),
                     client.send_queue().subscribe(),
-                    queues.upload_progress,
+                    queues.upload_reports,
                     events(),
                 )),
             ),
@@ -654,7 +654,7 @@ impl Engine {
         let (subscribed, ready) = watch::channel(false);
         self.tasks.spawn(
             TaskContext::Command("upload watcher"),
-            watchers::room_uploads(room.clone(), self.upload_progress.clone(), subscribed),
+            watchers::room_uploads(room.clone(), self.upload_reports.clone(), subscribed),
         );
         self.uploads
             .insert(room.room_id().to_owned(), ready.clone());

Modified file: src/engine/watchers.rs+143 additions−68 deletions@@ -145,10 +145,12 @@ pub(super) async fn send_queue_errors(
     }
 }
 
-/// upload progress only goes out on the room's own channel, never the client-wide one
+/// upload progress only goes out on the room's own channel, never the client-wide one.
+/// the room's sends and cancels come along behind it, so an upload's end can't
+/// overtake its own progress
 pub(super) async fn room_uploads(
     room: Room,
-    progress: mpsc::Sender<UploadProgress>,
+    tracker: mpsc::Sender<RoomUpload>,
     ready: watch::Sender<bool>,
 ) {
     let room_id = room.room_id().to_owned();
@@ -161,38 +163,49 @@ pub(super) async fn room_uploads(
     };
     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,
+        let report = match updates.recv().await {
+            Ok(RoomSendQueueUpdate::MediaUpload {
+                related_to,
+                file,
+                progress,
+                ..
+            }) => RoomUpload::Progress {
+                upload: (room_id.clone(), related_to),
+                uploaded_bytes: progress.current as u64,
+                total_bytes: progress.total as u64,
+                complete: file.is_some() && progress.current == progress.total,
+            },
+            Ok(RoomSendQueueUpdate::SentEvent {
+                transaction_id,
+                event_id,
+            }) => RoomUpload::Ended(
+                (room_id.clone(), transaction_id),
+                UploadOutcome::Sent { event_id },
+            ),
+            Ok(RoomSendQueueUpdate::CancelledLocalEvent { transaction_id }) => RoomUpload::Ended(
+                (room_id.clone(), transaction_id),
+                UploadOutcome::Cancelled,
+            ),
+            Ok(_) => continue,
+            Err(RecvError::Lagged(_)) => RoomUpload::Lagged,
             Err(RecvError::Closed) => return,
         };
-        let RoomSendQueueUpdate::MediaUpload {
-            related_to,
-            file,
-            progress: done,
-            ..
-        } = update
-        else {
-            continue;
-        };
-        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 progress.send(tick).await.is_err() {
+        if tracker.send(report).await.is_err() {
             return;
         }
     }
 }
 
-pub(super) struct UploadProgress {
-    upload: Upload,
-    uploaded_bytes: u64,
-    total_bytes: u64,
-    complete: bool,
+pub(super) enum RoomUpload {
+    Progress {
+        upload: Upload,
+        uploaded_bytes: u64,
+        total_bytes: u64,
+        complete: bool,
+    },
+    Ended(Upload, UploadOutcome),
+    /// an end may have been lost with the ticks
+    Lagged,
 }
 
 /// the one place upload events leave from, so an upload is never reported after it
@@ -201,7 +214,7 @@ pub(super) struct UploadProgress {
 pub(super) async fn uploads(
     client: Client,
     mut updates: tokio::sync::broadcast::Receiver<SendQueueUpdate>,
-    mut progress: mpsc::Receiver<UploadProgress>,
+    mut rooms_reports: mpsc::Receiver<RoomUpload>,
     events: EventSender,
 ) {
     let (_, mut rooms) = client.rooms_stream();
@@ -214,13 +227,15 @@ pub(super) async fn uploads(
             update = updates.recv() => match update {
                 Ok(update) => tracker.update(update).into_iter().collect(),
                 Err(RecvError::Lagged(skipped)) => {
-                    tracing::warn!(skipped, "upload watcher lagged");
-                    let current = attachment_echoes(&client).await;
-                    tracker.resnapshot(current)
+                    tracing::warn!(skipped, "upload tracker lagged");
+                    tracker.resnapshot(attachment_echoes(&client).await)
                 }
                 Err(RecvError::Closed) => return,
             },
-            Some(tick) = progress.recv() => tracker.progress(tick).into_iter().collect(),
+            Some(report) = rooms_reports.recv() => match report {
+                RoomUpload::Lagged => tracker.resnapshot(attachment_echoes(&client).await),
+                report => tracker.room(report).into_iter().collect(),
+            },
             update = rooms.next() => {
                 if update.is_none() {
                     return;
@@ -241,9 +256,9 @@ pub(super) async fn uploads(
 
 type Upload = (OwnedRoomId, OwnedTransactionId);
 
-/// 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
+/// everything reported gets exactly one `MediaUploadFinished`. an upload ended from
+/// anywhere but its room's forwarder goes in `ended` until the forwarder's own end for
+/// it arrives, since until then the forwarder can still be holding progress for it
 struct UploadTracker {
     live: HashSet<Upload>,
     ended: HashSet<Upload>,
@@ -286,22 +301,36 @@ impl UploadTracker {
             }
             _ => return None,
         };
-        self.finish((room_id, transaction_id), outcome)
+        self.finish((room_id, transaction_id), outcome, true)
     }
 
-    fn progress(&mut self, tick: UploadProgress) -> Option<Event> {
-        if self.ended.contains(&tick.upload) {
-            return None;
+    fn room(&mut self, report: RoomUpload) -> Option<Event> {
+        match report {
+            RoomUpload::Progress {
+                upload,
+                uploaded_bytes,
+                total_bytes,
+                complete,
+            } => {
+                if self.ended.contains(&upload) {
+                    return None;
+                }
+                self.live.insert(upload.clone());
+                let (room_id, transaction_id) = upload;
+                Some(Event::MediaUploadProgress {
+                    room_id,
+                    transaction_id,
+                    uploaded_bytes,
+                    total_bytes,
+                    complete,
+                })
+            }
+            RoomUpload::Ended(upload, outcome) => {
+                self.ended.remove(&upload);
+                self.finish(upload, outcome, false)
+            }
+            RoomUpload::Lagged => None,
         }
-        self.live.insert(tick.upload.clone());
-        let (room_id, transaction_id) = tick.upload;
-        Some(Event::MediaUploadProgress {
-            room_id,
-            transaction_id,
-            uploaded_bytes: tick.uploaded_bytes,
-            total_bytes: tick.total_bytes,
-            complete: tick.complete,
-        })
     }
 
     fn resnapshot(&mut self, current: HashSet<Upload>) -> Vec<Event> {
@@ -325,15 +354,17 @@ impl UploadTracker {
             .collect::<Vec<_>>();
         ending
             .into_iter()
-            .filter_map(|upload| self.finish(upload, outcome.clone()))
+            .filter_map(|upload| self.finish(upload, outcome.clone(), true))
             .collect()
     }
 
-    fn finish(&mut self, upload: Upload, outcome: UploadOutcome) -> Option<Event> {
+    fn finish(&mut self, upload: Upload, outcome: UploadOutcome, remember: bool) -> Option<Event> {
         if !self.live.remove(&upload) {
             return None;
         }
-        self.ended.insert(upload.clone());
+        if remember {
+            self.ended.insert(upload.clone());
+        }
         let (room_id, transaction_id) = upload;
         Some(Event::MediaUploadFinished {
             room_id,
@@ -445,27 +476,49 @@ mod tests {
     use matrix_sdk::send_queue::{RoomSendQueueUpdate, SendQueueUpdate};
     use ruma::{OwnedTransactionId, event_id, owned_room_id};
 
-    use super::{HashSet, UploadProgress, UploadTracker};
+    use super::{HashSet, RoomUpload, Upload, UploadTracker};
     use crate::{Event, UploadOutcome};
 
-    #[test]
-    fn an_upload_only_progress_saw_still_ends_once() {
-        let upload = (
+    fn upload() -> Upload {
+        (
             owned_room_id!("!room:localhost"),
             OwnedTransactionId::from("upload"),
-        );
-        let tick = || UploadProgress {
-            upload: upload.clone(),
+        )
+    }
+
+    fn tick() -> RoomUpload {
+        RoomUpload::Progress {
+            upload: upload(),
             uploaded_bytes: 1,
             total_bytes: 2,
             complete: false,
-        };
-        let mut tracker = UploadTracker {
+        }
+    }
+
+    fn sent() -> UploadOutcome {
+        UploadOutcome::Sent {
+            event_id: event_id!("$sent:localhost").to_owned(),
+        }
+    }
+
+    fn tracker() -> UploadTracker {
+        UploadTracker {
             live: HashSet::new(),
             ended: HashSet::new(),
-        };
-        assert!(tracker.progress(tick()).is_some());
+        }
+    }
+
+    fn finished(event: Option<Event>) -> Option<UploadOutcome> {
+        match event? {
+            Event::MediaUploadFinished { outcome, .. } => Some(outcome),
+            _ => None,
+        }
+    }
 
+    #[test]
+    fn a_stale_tick_after_a_missed_end_changes_nothing() {
+        let mut tracker = tracker();
+        assert!(tracker.room(tick()).is_some());
         let missed = tracker.resnapshot(HashSet::new());
         assert!(matches!(
             missed.as_slice(),
@@ -475,15 +528,37 @@ mod tests {
             }]
         ));
 
-        assert!(tracker.progress(tick()).is_none());
-        let sent = SendQueueUpdate {
-            room_id: upload.0.clone(),
+        assert!(tracker.room(tick()).is_none());
+        let (room_id, transaction_id) = upload();
+        let global = SendQueueUpdate {
+            room_id,
+            update: RoomSendQueueUpdate::SentEvent {
+                transaction_id,
+                event_id: event_id!("$sent:localhost").to_owned(),
+            },
+        };
+        assert!(tracker.update(global).is_none());
+        assert!(tracker.room(RoomUpload::Ended(upload(), sent())).is_none());
+        assert!(tracker.live.is_empty() && tracker.ended.is_empty());
+    }
+
+    #[test]
+    fn an_end_the_tracker_never_knew_about_still_follows_its_progress() {
+        let mut tracker = tracker();
+        let (room_id, transaction_id) = upload();
+        let global = SendQueueUpdate {
+            room_id,
             update: RoomSendQueueUpdate::SentEvent {
-                transaction_id: upload.1.clone(),
+                transaction_id,
                 event_id: event_id!("$sent:localhost").to_owned(),
             },
         };
-        assert!(tracker.update(sent).is_none());
-        assert!(tracker.resnapshot(HashSet::from([upload])).is_empty());
+        assert!(tracker.update(global).is_none());
+        assert!(tracker.room(tick()).is_some());
+        assert_eq!(
+            finished(tracker.room(RoomUpload::Ended(upload(), sent()))),
+            Some(sent())
+        );
+        assert!(tracker.live.is_empty() && tracker.ended.is_empty());
     }
 }

Modified file: src/model.rs+3 additions−1 deletions@@ -681,7 +681,9 @@ pub enum Event {
         kind: crate::ErrorKind,
         will_retry: bool,
     },
-    /// every upload ends with exactly one of these
+    /// an upload that turned up in a progress or failure event gets exactly one of
+    /// these while the engine runs, and nothing about it after. stopping ends
+    /// nothing, `Stopped` covers whatever is still going and the sdk resumes it next run
     MediaUploadFinished {
         room_id: OwnedRoomId,
         transaction_id: OwnedTransactionId,

macltok

Commit 7cf348521204

Commit `7cf348521204`