macltok

public

Star0

Fork 0

Watch0

9 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `0bfa3f46f0be`

tarball

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

9 files changed+195-56

engine: say when timelines, typing and uploads end

A consumer keeping state per timeline, per room's typing or per upload had no event for when to drop it, so closes, leaves and cancelled uploads left it guessing. The upload watcher now follows every attachment in the send queue itself rather than asking the room's queue on each failure, because the client-wide channel carries sends and cancels as well and a failed upload can still be retried. Push rule changes don't raise a room info update, so NotificationSettingsChanged now brings a RoomsChanged for the notification modes in RoomSummary.

Modified file: docs/design-decisions.md+14 additions−3 deletions@@ -58,6 +58,10 @@ kicked or banned from, closes itself. it tells the engine first, and anything
 already queued to it comes back to be answered or replayed, so an open racing a
 leave still gets a fresh actor
 
+however an actor ends, short of the engine stopping, its typing goes out empty
+and each of its timelines sends `TimelineClosed`. so does a closed thread. a
+timeline that never sent a reset never sends a close either
+
 opening sends `TimelineReset` on the stream before it answers. opening an
 already open room sends a fresh reset, so a second frontend gets a snapshot
 
@@ -138,8 +142,12 @@ closes, least recently seen first once the cap is hit
 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. failures do reach the client-wide channel, so one
-watcher there reports them
+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
+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
 
 ## sessions and secrets
 
@@ -175,7 +183,10 @@ sdk ui types stay behind the boundary
 
 `rooms` normalises on demand with bounded concurrency and sorts by latest event,
 then name, then id. `RoomsChanged` is an invalidation, coalesced so one sync
-touching fifty rooms is one event
+touching fifty rooms is one event. a left room stays in the list with its
+membership until it's forgotten. push rule changes never touch room info, so
+`NotificationSettingsChanged` brings a `RoomsChanged` along for the modes in
+the summaries
 
 ## verification
 

Modified file: src/engine/mod.rs+2 additions−2 deletions@@ -401,8 +401,8 @@ impl Engine {
         let events = || self.port.event_sender();
         let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 11] = [
             (
-                "upload failure watcher",
-                Box::pin(watchers::upload_failures(
+                "upload watcher",
+                Box::pin(watchers::uploads(
                     client.clone(),
                     client.send_queue().subscribe(),
                     events(),

Modified file: src/engine/room.rs+11 additions−2 deletions@@ -91,7 +91,7 @@ impl RoomActor {
             reply.done(Ok(()));
         }
         let (typing_guard, typing) = self.room.subscribe_to_typing_notifications();
-        let typing_watcher = TaskGuard(self.context.supervisor.spawn(
+        let mut typing_watcher = TaskGuard(self.context.supervisor.spawn(
             "typing watcher",
             watch_typing(self.room_id.clone(), typing, self.context.events.clone()),
         ));
@@ -112,7 +112,16 @@ impl RoomActor {
         // let redactions and receipts that already left finish rather than lie about them
         while timelines.work.join_next().await.is_some() {}
         drop(typing_guard);
-        drop(typing_watcher);
+        typing_watcher.0.abort();
+        let _ = (&mut typing_watcher.0).await;
+        let _ = self
+            .context
+            .events
+            .send(Event::TypingChanged(TypingState {
+                room_id: self.room_id.clone(),
+                users: Vec::new(),
+            }))
+            .await;
         timelines.threads.close().await;
         timelines.room.close().await;
         if let Some(reply) = close {

Modified file: src/engine/timeline.rs+7 additions@@ -159,6 +159,13 @@ impl TimelineSession {
         let _ = (&mut self.watcher).await;
         // after the watcher, or a last diff could register media for a timeline that's gone
         self.context.media.release(&self.scope).await;
+        let _ = self
+            .context
+            .events
+            .send(Event::TimelineClosed {
+                timeline_id: self.id.clone(),
+            })
+            .await;
     }
 
     pub(super) async fn execute(

Modified file: src/engine/watchers.rs+117 additions−45 deletions@@ -1,19 +1,19 @@
-use std::{sync::Arc, time::Duration};
+use std::{collections::HashSet, sync::Arc, time::Duration};
 
 use futures_util::StreamExt;
 use matrix_sdk::{
     Client, Room, SessionChange,
     ruma::events::{AnyMessageLikeEventContent, room::message::MessageType},
-    send_queue::{LocalEchoContent, RoomSendQueueUpdate},
+    send_queue::{LocalEchoContent, RoomSendQueueUpdate, SendQueueUpdate},
 };
 use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};
-use ruma::{OwnedDeviceId, time::Instant};
+use ruma::{OwnedDeviceId, OwnedRoomId, OwnedTransactionId, 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,
+    ConnectionState, Event, Termination, UploadOutcome, encryption::encryption_state,
+    error::classify_sdk_error, transport::EventSender,
 };
 
 const SYNC_RESTART_MIN: Duration = Duration::from_secs(1);
@@ -185,54 +185,127 @@ pub(super) async fn room_uploads(room: Room, events: EventSender, ready: watch::
     }
 }
 
-/// failures do reach the client-wide channel, and this one exists before any upload does
-pub(super) async fn upload_failures(
+/// progress only goes out on each room's own channel, everything else reaches this
+/// one, which exists before any upload does
+pub(super) async fn uploads(
     client: Client,
-    mut updates: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueUpdate>,
+    mut updates: tokio::sync::broadcast::Receiver<SendQueueUpdate>,
     events: EventSender,
 ) {
+    let (_, mut rooms) = client.rooms_stream();
+    let mut uploads = attachment_echoes(&client).await;
     loop {
-        let update = match updates.recv().await {
-            Ok(update) => update,
-            Err(RecvError::Lagged(skipped)) => {
-                tracing::warn!(skipped, "upload failure watcher lagged");
-                continue;
+        let finished = tokio::select! {
+            update = updates.recv() => match update {
+                Ok(update) => match upload_update(&mut uploads, update) {
+                    Some(event) => vec![event],
+                    None => continue,
+                },
+                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)
+                }
+                Err(RecvError::Closed) => return,
+            },
+            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)
             }
-            Err(RecvError::Closed) => return,
         };
-        let RoomSendQueueUpdate::SendError {
+        for event in finished {
+            if events.send(event).await.is_err() {
+                return;
+            }
+        }
+    }
+}
+
+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()
+}
+
+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,
-        } = 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)
-                })
+        } => {
+            let upload = (room_id, transaction_id);
+            if !uploads.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,
             });
-        if !is_attachment {
-            continue;
         }
-        let event = Event::MediaUploadFailed {
-            room_id: update.room_id,
+        RoomSendQueueUpdate::SentEvent {
             transaction_id,
-            kind: classify_sdk_error(&error),
-            recoverable: is_recoverable,
-        };
-        if events.send(event).await.is_err() {
-            return;
+            event_id,
+        } => (transaction_id, UploadOutcome::Sent { event_id }),
+        RoomSendQueueUpdate::CancelledLocalEvent { transaction_id } => {
+            (transaction_id, UploadOutcome::Cancelled)
         }
+        _ => return None,
+    };
+    let upload = (room_id, transaction_id);
+    if !uploads.remove(&upload) {
+        return None;
     }
+    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
+        .into_iter()
+        .map(|(room_id, transaction_id)| Event::MediaUploadFinished {
+            room_id,
+            transaction_id,
+            outcome: outcome.clone(),
+        })
+        .collect()
 }
 
 fn is_attachment_echo(content: &LocalEchoContent) -> bool {
@@ -308,12 +381,11 @@ 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;
+        // room summaries carry the effective mode, and push rule changes never touch room info
+        for event in [Event::NotificationSettingsChanged, Event::RoomsChanged] {
+            if events.send(event).await.is_err() {
+                return;
+            }
         }
     }
 }

Modified file: src/lib.rs+2 additions−2 deletions@@ -38,8 +38,8 @@ pub use model::{
     RoomNotificationSettings, RoomPowerLevels, RoomPreset, RoomSummary, RoomVisibility, SendState,
     Sender, SpaceChild, Termination, ThreadSummary, TimelineCommand, TimelineContent, TimelineDiff,
     TimelineDiffError, TimelineEvent, TimelineId, TimelineItem, TimelineItemId, TypingState,
-    UploadDetails, UploadThumbnail, UtdCause, VerificationAction, VerificationFailure,
-    VerificationState, VerificationTarget,
+    UploadDetails, UploadOutcome, UploadThumbnail, UtdCause, VerificationAction,
+    VerificationFailure, VerificationState, VerificationTarget,
 };
 pub use transport::{ClientHandle, EventStream, Pending, SpaceHierarchy};
 pub use uiaa::{

Modified file: src/model.rs+29 additions−1 deletions@@ -658,6 +658,12 @@ pub enum Event {
         timeline_id: TimelineId,
         state: HistoryState,
     },
+    /// the timeline was closed, or its room was left, and nothing more comes for it
+    /// until another `TimelineReset`
+    TimelineClosed {
+        timeline_id: TimelineId,
+    },
+    /// a room that stops being watched says so with an empty `users`
     TypingChanged(TypingState),
     MediaUploadProgress {
         room_id: OwnedRoomId,
@@ -666,11 +672,19 @@ pub enum Event {
         total_bytes: u64,
         complete: bool,
     },
+    /// not the end of it. `will_retry` means the send queue has another go by itself,
+    /// otherwise it waits for `retry_send` or `cancel_send`
     MediaUploadFailed {
         room_id: OwnedRoomId,
         transaction_id: OwnedTransactionId,
         kind: crate::ErrorKind,
-        recoverable: bool,
+        will_retry: bool,
+    },
+    /// every upload ends with exactly one of these
+    MediaUploadFinished {
+        room_id: OwnedRoomId,
+        transaction_id: OwnedTransactionId,
+        outcome: UploadOutcome,
     },
     EncryptionChanged(EncryptionState),
     VerificationChanged(VerificationState),
@@ -682,6 +696,20 @@ pub enum Event {
     Stopped,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub enum UploadOutcome {
+    Sent {
+        event_id: OwnedEventId,
+    },
+    Cancelled,
+    /// its room was forgotten, and the queue with it
+    RoomForgotten,
+    /// the sdk's update channel overflowed while it ended, so whether it was sent is unknown
+    Missed,
+}
+
 /// why `Engine::run` came back, when nothing went wrong
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]

Modified file: tests/engine.rs+4 additions@@ -180,6 +180,10 @@ async fn keeps_open_room_timelines_independent() {
     .expect("both retained rooms did not update");
 
     handle.close_room(first_id.to_owned()).await.unwrap();
+    receive_until(&mut events, |event| {
+        matches!(event, Event::TimelineClosed { timeline_id } if timeline_id.room_id() == first_id)
+    })
+    .await;
     server
         .sync_room(
             &sync_client,

Modified file: tests/media.rs+9 additions−1 deletions@@ -4,7 +4,7 @@ use std::{sync::Arc, time::Duration};
 
 use macltok::{
     AttachmentUpload, ClientHandle, Engine, EngineConfig, EngineError, ErrorKind, Event,
-    EventStream, MediaId, MediaSize, UploadDetails, UploadThumbnail,
+    EventStream, MediaId, MediaSize, UploadDetails, UploadOutcome, UploadThumbnail,
 };
 use matrix_sdk::{
     media::{MediaFormat, MediaRequestParameters},
@@ -600,6 +600,14 @@ async fn voice_upload_emits_voice_details_and_correlated_progress() {
         total_bytes,
         complete: true,
     } if room_id == room_id!("!media:localhost") && seen == transaction_id && uploaded_bytes == data.len() as u64 && total_bytes == data.len() as u64));
+    support::receive_until(&mut harness.events, |event| {
+        matches!(event, Event::MediaUploadFinished {
+            transaction_id: seen,
+            outcome: UploadOutcome::Sent { event_id },
+            ..
+        } if seen == &transaction_id && event_id == event_id!("$voice-upload:localhost"))
+    })
+    .await;
 
     let content = uploaded_content(&harness.server).await;
     assert_eq!(content["msgtype"], "m.audio");

macltok

Commit 0bfa3f46f0be

Commit `0bfa3f46f0be`