macltok

public

Star0

Fork 0

Watch0

39 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `411bc1c78b65`

tarball

atagen <boss@atagen.co>2026-10-22 09:31 UTC

4 files changed+157-227

uploads: follow progress on the client-wide channel and drop the per-room forwarders

Upstream only sends upload progress on each room's own channel, which needed a forwarder per uploading room, a readiness handshake before queueing, restarts for resumed uploads, and a tracker reconciling two channels' views of the same upload. The fork's enable_global_upload_progress puts it all on one channel. The only ordering left to handle is the sdk's detached progress task trailing an upload's end.

Modified file: docs/design-decisions.md+12 additions−13 deletions@@ -188,23 +188,20 @@ memory, the very thing the host streams to avoid. `media_source` hands over the
 url and keys `media` would have used, picked the same way, and `access_token`
 the rest. the byte cap doesn't apply, nothing is buffered
 
-uploads are commands that answer once the send queue has them. progress only
-goes out on a room's own send queue channel, so each room that uploads gets a
-forwarder, subscribed before the upload is queued and restarted for uploads the
-sdk resumes from last time. everything but progress reaches the client-wide
-channel, so one watcher there follows every attachment in the queue and ends
-each with one `MediaUploadFinished`. sent, cancelled, its room forgotten, or
+uploads are commands that answer once the send queue has them. the fork puts
+upload progress on the client-wide send queue channel too, which upstream keeps
+for each room's own, so one watcher there follows every attachment in the queue,
+leftovers the sdk resumes from last time included, and ends each with one
+`MediaUploadFinished`. sent, cancelled, its room forgotten, or
 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. 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
+that watcher is 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. the sdk sends
+progress from a detached task, so a tick can turn up behind its own upload's
+end, and the watcher remembers the last few hundred ends to drop those. the
 engine stopping ends nothing, `Stopped` covers it
 
 ## sessions and secrets
@@ -390,6 +387,8 @@ which keep their public sans-io apis
   sdk's hard-coded list leaves out. image packs use it
 - msc4027 reaction shortcodes, kept on the timeline's `ReactionInfo` and sent
   with `Timeline::toggle_reaction_with_shortcode`
+- `SendQueue::enable_global_upload_progress`, upload progress on the
+  client-wide send queue channel
 
 ## public model
 

Modified file: src/engine/mod.rs+1 additions−44 deletions@@ -246,7 +246,6 @@ struct Queues {
     encryption: mpsc::Receiver<Job<EncryptionCommand>>,
     session_changes: broadcast::Receiver<SessionChange>,
     persistence: Option<Persistence>,
-    upload_reports: mpsc::Receiver<watchers::RoomUpload>,
 }
 
 /// requests other devices send us, handed from the sdk's event handler to the encryption worker
@@ -270,9 +269,6 @@ pub struct Engine {
     notifications: mpsc::Sender<Job<NotificationCommand>>,
     encryption: mpsc::Sender<Job<EncryptionCommand>>,
     queues: Option<Queues>,
-    /// one progress forwarder per room that has uploaded, and whether it's subscribed yet
-    uploads: HashMap<OwnedRoomId, watch::Receiver<bool>>,
-    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>,
@@ -334,9 +330,9 @@ impl Engine {
             config.event_capacity,
         );
         client.send_queue().enable_upload_progress(true);
+        client.send_queue().enable_global_upload_progress(true);
         let (note_sender, notes) = mpsc::unbounded_channel();
         let (stop, stops) = mpsc::unbounded_channel();
-        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);
@@ -363,10 +359,7 @@ impl Engine {
                 encryption: encryption_queue,
                 session_changes,
                 persistence,
-                upload_reports: upload_report_queue,
             }),
-            upload_reports,
-            uploads: HashMap::new(),
             verification_handler: None,
             preparations: prepare::Preparations::default(),
             stop,
@@ -403,7 +396,6 @@ impl Engine {
             }
         }
         self.start_tasks();
-        self.resume_upload_forwarders().await;
         self.sync.start().await;
 
         let mut shutdown_reply = None;
@@ -470,7 +462,6 @@ impl Engine {
                 Box::pin(watchers::uploads(
                     client.clone(),
                     client.send_queue().subscribe(),
-                    queues.upload_reports,
                     events(),
                 )),
             ),
@@ -740,45 +731,12 @@ impl Engine {
         let Some(room) = self.client.get_room(upload.timeline_id.room_id()) else {
             return reply.done(Err(Error::room_unknown()));
         };
-        let mut ready = self.upload_forwarder(&room);
         let max_bytes = self.config.max_media_bytes;
-        // queueing before the forwarder listens would lose the first progress ticks
         self.spawn_command("upload", reply, async move {
-            ready
-                .wait_for(|subscribed| *subscribed)
-                .await
-                .map_err(|_| Error::new(ErrorKind::Store, "reading the media send queue failed"))?;
             media::queue_upload(room, upload, max_bytes).await
         });
     }
 
-    fn upload_forwarder(&mut self, room: &matrix_sdk::Room) -> watch::Receiver<bool> {
-        self.uploads.retain(|_, ready| ready.has_changed().is_ok());
-        if let Some(ready) = self.uploads.get(room.room_id()) {
-            return ready.clone();
-        }
-        let (subscribed, ready) = watch::channel(false);
-        self.tasks.spawn(
-            TaskContext::Command("upload watcher"),
-            watchers::room_uploads(room.clone(), self.upload_reports.clone(), subscribed),
-        );
-        self.uploads
-            .insert(room.room_id().to_owned(), ready.clone());
-        ready
-    }
-
-    /// the sdk resumes uploads left over from last time, and they deserve progress too
-    async fn resume_upload_forwarders(&mut self) {
-        let Ok(echoes) = self.client.send_queue().local_echoes().await else {
-            return;
-        };
-        for room_id in echoes.keys() {
-            if let Some(room) = self.client.get_room(room_id) {
-                self.upload_forwarder(&room);
-            }
-        }
-    }
-
     fn route_timeline(&mut self, command: TimelineCommand, reply: Responder) {
         let sender = match self.rooms.get(command.room_id()) {
             Some(RoomSlot::Active { sender, .. }) => Some(sender.clone()),
@@ -948,7 +906,6 @@ impl Engine {
         self.port.close();
         self.tasks.shutdown().await;
         self.rooms.clear();
-        self.uploads.clear();
         self.media.clear().await;
         self.sync.stop().await;
         self.port

Modified file: src/engine/watchers.rs+140 additions−170 deletions@@ -1,8 +1,12 @@
-use std::{collections::HashSet, sync::Arc, time::Duration};
+use std::{
+    collections::{HashSet, VecDeque},
+    sync::Arc,
+    time::Duration,
+};
 
 use futures_util::StreamExt;
 use matrix_sdk::{
-    Client, Room,
+    Client,
     ruma::events::{AnyMessageLikeEventContent, room::message::MessageType},
     send_queue::{LocalEchoContent, RoomSendQueueUpdate, SendQueueUpdate},
 };
@@ -14,7 +18,7 @@ use ruma::{
     serde::Raw,
     time::Instant,
 };
-use tokio::sync::{Notify, broadcast::error::RecvError, mpsc, watch};
+use tokio::sync::{Notify, broadcast::error::RecvError, watch};
 
 use super::executor::TaskSet;
 use crate::{
@@ -122,84 +126,20 @@ pub(super) async fn send_queue_errors(
     }
 }
 
-/// 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,
-    tracker: mpsc::Sender<RoomUpload>,
-    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 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,
-        };
-        if tracker.send(report).await.is_err() {
-            return;
-        }
-    }
-}
-
-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
-/// ended. progress comes in from the room forwarders, the rest from the client-wide
-/// channel, which exists before any upload does
+/// ended. the client-wide channel exists before any upload does. it holds 32 updates and
+/// progress ticks share it, so this keeps draining while the event stream is full
 pub(super) async fn uploads(
     client: Client,
     mut updates: tokio::sync::broadcast::Receiver<SendQueueUpdate>,
-    mut rooms_reports: mpsc::Receiver<RoomUpload>,
     events: EventSender,
 ) {
     let (_, mut rooms) = client.rooms_stream();
-    let mut tracker = UploadTracker {
-        live: attachment_echoes(&client).await,
-        ended: HashSet::new(),
-    };
+    let mut tracker = UploadTracker::new(attachment_echoes(&client).await);
+    let mut pending = VecDeque::new();
     loop {
-        let reports = tokio::select! {
+        let reports: Vec<Event> = tokio::select! {
+            biased;
             update = updates.recv() => match update {
                 Ok(update) => tracker.update(update).into_iter().collect(),
                 Err(RecvError::Lagged(skipped)) => {
@@ -208,10 +148,6 @@ pub(super) async fn uploads(
                 }
                 Err(RecvError::Closed) => return,
             },
-            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;
@@ -221,26 +157,77 @@ pub(super) async fn uploads(
                     &UploadOutcome::RoomForgotten,
                 )
             }
+            permit = events.reserve(), if !pending.is_empty() => {
+                let Ok(permit) = permit else { return };
+                permit.send(pending.pop_front().expect("pending was checked"));
+                continue;
+            }
         };
         for event in reports {
-            if events.send(event).await.is_err() {
-                return;
-            }
+            queue_upload_event(&mut pending, event);
+        }
+    }
+}
+
+/// a progress tick replaces a waiting one for the same upload, unless something else about
+/// that upload is queued after it
+fn queue_upload_event(pending: &mut VecDeque<Event>, event: Event) {
+    if matches!(event, Event::MediaUploadProgress { .. })
+        && let Some(waiting) = pending
+            .iter_mut()
+            .rev()
+            .find(|waiting| upload_of(waiting) == upload_of(&event))
+        && matches!(waiting, Event::MediaUploadProgress { .. })
+    {
+        *waiting = event;
+        return;
+    }
+    pending.push_back(event);
+}
+
+fn upload_of(event: &Event) -> Option<(&OwnedRoomId, &OwnedTransactionId)> {
+    match event {
+        Event::MediaUploadProgress {
+            room_id,
+            transaction_id,
+            ..
+        }
+        | Event::MediaUploadFailed {
+            room_id,
+            transaction_id,
+            ..
         }
+        | Event::MediaUploadFinished {
+            room_id,
+            transaction_id,
+            ..
+        } => Some((room_id, transaction_id)),
+        _ => None,
     }
 }
 
 type Upload = (OwnedRoomId, OwnedTransactionId);
 
-/// 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
+/// the sdk sends progress ticks from a detached task, so a tick can land after its
+/// upload's end. it trails by one task hop, so a short memory of ends covers it
+const REMEMBERED_ENDS: usize = 256;
+
+/// everything reported gets exactly one `MediaUploadFinished` and nothing after it
 struct UploadTracker {
     live: HashSet<Upload>,
     ended: HashSet<Upload>,
+    ended_order: VecDeque<Upload>,
 }
 
 impl UploadTracker {
+    fn new(live: HashSet<Upload>) -> Self {
+        Self {
+            live,
+            ended: HashSet::new(),
+            ended_order: VecDeque::new(),
+        }
+    }
+
     fn update(&mut self, update: SendQueueUpdate) -> Option<Event> {
         let room_id = update.room_id;
         let (transaction_id, outcome) = match update.update {
@@ -251,6 +238,26 @@ impl UploadTracker {
                 }
                 return None;
             }
+            RoomSendQueueUpdate::MediaUpload {
+                related_to,
+                file,
+                progress,
+                ..
+            } => {
+                let upload = (room_id, related_to);
+                if self.ended.contains(&upload) {
+                    return None;
+                }
+                self.live.insert(upload.clone());
+                let (room_id, transaction_id) = upload;
+                return Some(Event::MediaUploadProgress {
+                    room_id,
+                    transaction_id,
+                    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,
@@ -277,36 +284,7 @@ impl UploadTracker {
             }
             _ => return None,
         };
-        self.finish((room_id, transaction_id), outcome, true)
-    }
-
-    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.finish((room_id, transaction_id), outcome)
     }
 
     fn resnapshot(&mut self, current: HashSet<Upload>) -> Vec<Event> {
@@ -330,17 +308,23 @@ impl UploadTracker {
             .collect::<Vec<_>>();
         ending
             .into_iter()
-            .filter_map(|upload| self.finish(upload, outcome.clone(), true))
+            .filter_map(|upload| self.finish(upload, outcome.clone()))
             .collect()
     }
 
-    fn finish(&mut self, upload: Upload, outcome: UploadOutcome, remember: bool) -> Option<Event> {
+    /// an end for an upload we never saw is remembered too, its ticks may still be coming
+    fn finish(&mut self, upload: Upload, outcome: UploadOutcome) -> Option<Event> {
+        if self.ended.insert(upload.clone()) {
+            self.ended_order.push_back(upload.clone());
+            if self.ended_order.len() > REMEMBERED_ENDS
+                && let Some(oldest) = self.ended_order.pop_front()
+            {
+                self.ended.remove(&oldest);
+            }
+        }
         if !self.live.remove(&upload) {
             return None;
         }
-        if remember {
-            self.ended.insert(upload.clone());
-        }
         let (room_id, transaction_id) = upload;
         Some(Event::MediaUploadFinished {
             room_id,
@@ -504,10 +488,10 @@ pub(super) async fn notification_settings(client: Client, events: EventSender) {
 
 #[cfg(test)]
 mod tests {
-    use matrix_sdk::send_queue::{RoomSendQueueUpdate, SendQueueUpdate};
+    use matrix_sdk::send_queue::{AbstractProgress, RoomSendQueueUpdate, SendQueueUpdate};
     use ruma::{OwnedTransactionId, event_id, owned_room_id};
 
-    use super::{HashSet, RoomUpload, Upload, UploadTracker};
+    use super::{HashSet, Upload, UploadTracker};
     use crate::{Event, UploadOutcome};
 
     fn upload() -> Upload {
@@ -517,26 +501,30 @@ mod tests {
         )
     }
 
-    fn tick() -> RoomUpload {
-        RoomUpload::Progress {
-            upload: upload(),
-            uploaded_bytes: 1,
-            total_bytes: 2,
-            complete: false,
+    fn update(update: RoomSendQueueUpdate) -> SendQueueUpdate {
+        SendQueueUpdate {
+            room_id: upload().0,
+            update,
         }
     }
 
-    fn sent() -> UploadOutcome {
-        UploadOutcome::Sent {
-            event_id: event_id!("$sent:localhost").to_owned(),
-        }
+    fn tick() -> SendQueueUpdate {
+        update(RoomSendQueueUpdate::MediaUpload {
+            related_to: upload().1,
+            file: None,
+            index: 0,
+            progress: AbstractProgress {
+                current: 1,
+                total: 2,
+            },
+        })
     }
 
-    fn tracker() -> UploadTracker {
-        UploadTracker {
-            live: HashSet::new(),
-            ended: HashSet::new(),
-        }
+    fn sent() -> SendQueueUpdate {
+        update(RoomSendQueueUpdate::SentEvent {
+            transaction_id: upload().1,
+            event_id: event_id!("$sent:localhost").to_owned(),
+        })
     }
 
     fn finished(event: Option<Event>) -> Option<UploadOutcome> {
@@ -547,9 +535,21 @@ mod tests {
     }
 
     #[test]
+    fn a_tick_trailing_its_end_says_nothing() {
+        let mut tracker = UploadTracker::new(HashSet::new());
+        assert!(tracker.update(tick()).is_some());
+        assert!(matches!(
+            finished(tracker.update(sent())),
+            Some(UploadOutcome::Sent { .. })
+        ));
+        assert!(tracker.update(tick()).is_none());
+        assert!(tracker.live.is_empty());
+    }
+
+    #[test]
     fn a_stale_tick_after_a_missed_end_changes_nothing() {
-        let mut tracker = tracker();
-        assert!(tracker.room(tick()).is_some());
+        let mut tracker = UploadTracker::new(HashSet::new());
+        assert!(tracker.update(tick()).is_some());
         let missed = tracker.resnapshot(HashSet::new());
         assert!(matches!(
             missed.as_slice(),
@@ -558,38 +558,8 @@ mod tests {
                 ..
             }]
         ));
-
-        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,
-                event_id: event_id!("$sent:localhost").to_owned(),
-            },
-        };
-        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());
+        assert!(tracker.update(tick()).is_none());
+        assert!(tracker.update(sent()).is_none());
+        assert!(tracker.live.is_empty());
     }
 }

Modified file: src/transport.rs+4 additions@@ -992,6 +992,10 @@ impl EventSender {
         self.0.send(event).await.map_err(|_| SendEventError)
     }
 
+    pub(crate) async fn reserve(&self) -> Result<mpsc::Permit<'_, Event>, SendEventError> {
+        self.0.reserve().await.map_err(|_| SendEventError)
+    }
+
     /// resolves once every `EventStream` is gone
     pub(crate) async fn closed(&self) {
         self.0.closed().await;

macltok

Commit 411bc1c78b65

Commit `411bc1c78b65`