macltok

public

Star0

Fork 0

Watch0

5 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `5a52012c9b1d`

tarball

atagen <boss@atagen.co>2026-09-25 01:04 UTC

21 files changed+908-342

engine: fix what the second review found in the rework

Three parallel reviews of the new engine found closes and opens racing a self-closing actor answering Interrupted, verification watchers subscribing after the state they needed, and a plain event able to plant an encrypted file's ciphertext in the shared cache. SendState::Failed { retriable } had the SDK's inverted meaning for a UI, and downloads died on the SDK client's 30 second timeout. Incoming verification comes over from Malice's vendored copy, thumbnails the server refuses fall back to the file, and Engine::run returns a boxed future because the deeper actor tripped the query depth limit in every consuming crate.

Modified file: Cargo.lock+1

@@ -1830,6 +1830,7 @@ dependencies = [

18301830 "ruma",

18311831 "serde",

18321832 "serde_json",

1833+ "sha2",

18331834 "thiserror 2.0.20",

18341835 "tokio",

18351836 "tracing",

Modified file: Cargo.toml+1

@@ -20,6 +20,7 @@ ruma = { version = "0.16.0", default-features = false }

2020 serde = { version = "1.0.219", features = ["derive", "rc"], optional = true }

2121 thiserror = "2.0.16"

2222 mime = "0.3.17"

23+sha2 = "0.10.9"

2324 tokio = { version = "1.47.1", features = ["rt", "sync", "time"] }

2425 tracing = "0.1.41"

2526 url = { version = "2.5.7", optional = true }

Modified file: docs/design-decisions.md+36 additions−12 deletions@@ -33,20 +33,27 @@ and gets logged. a watcher or a room actor that panics takes the engine down,
 emitting `Failed` then `Stopped`
 
 sync restarts itself with backoff after an error or an unexpected stop, and
-`ConnectionChanged` shows `Offline` meanwhile. every return to `Online` re-arms
+`ConnectionChanged` shows `Offline` meanwhile. a run that drops again inside
+thirty seconds counts as a failure too, or a server that answers `/versions`
+but refuses sync would bounce us between online and offline forever. every return to `Online` re-arms
 the send queues, since the sdk parks a room's queue on any send failure and
 never unparks it. a recoverable failure while online gets the room another go
 after a few seconds
 
 logout and a rejected token both stop the engine. `Shutdown` answers once
-`Stopped` has gone out
+`Stopped` has gone out, every one of them if several were queued
+
+`Engine::run` hands back a boxed future. the real one nests deep enough that
+every crate spawning it would trip the compiler's query depth
 
 ## rooms
 
 one actor per open room. the engine holds a slot per room, `Active` with a
 sender or `Closing` with a fifo of opens and closes that arrived while it wound
 down. those replay in order once the actor exits. a room you leave, or get
-kicked or banned from, closes itself
+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
 
 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
@@ -63,8 +70,9 @@ pagination runs one request at a time per timeline and emits `HistoryChanged`
 around each. the sdk remembers where history starts, we don't, because a gappy
 sync can unload it again
 
-every task inside an actor is supervised, and a panic in any of them panics the
-actor
+the actor's watchers are supervised, and a panic in any of them panics the
+actor. its command work isn't, a panicking redaction just answers `Interrupted`
+and gets logged
 
 ## timelines
 
@@ -82,6 +90,9 @@ changes and the common room state changes get their own variants, and anything
 else is `OtherState` or `Other` with its type. nothing is flattened into english
 in a body field
 
+thread receipts for a thread that isn't open go straight to the room, so
+marking a thread read works the same whether its room is on screen or not
+
 replies carry the replied-to event once it's loaded. the watcher asks the sdk
 for any it hasn't got, once per event, since the sdk won't on its own
 
@@ -106,17 +117,26 @@ sdk's own patcher is private
 
 media is a read. `ClientHandle::media` takes a `MediaId` a retained timeline
 handed you, `mxc_media` takes any plain mxc uri. both share one bounded
-download path with the byte cap the sdk's own download lacks, and
-concurrent asks for the same bytes share one flight
+download path with the byte cap the sdk's own download lacks, an hour's total
+deadline instead of the sdk client's thirty seconds, and a minute's patience
+for a stalled stream. concurrent asks for the same bytes share one flight.
+a thumbnail the server won't make falls back to the file itself.
+`MediaId` parses from a string, since a web client has to round trip it
 
 the cache holds plaintext, keyed like the sdk keys it, so the sdk and macltok
-can read each other's entries. that's how your own encrypted uploads come back
+can read each other's entries. that's how your own encrypted uploads come back.
+the sdk keys plain and encrypted sources alike, so a plain event pointing at
+someone's encrypted file can leave ciphertext in the cache. an encrypted read
+checks the cached bytes against the file's hash and decrypts them if they match
 
 media ids are registered as timelines render and forgotten when their timeline
 closes, least recently seen first once the cap is hit
 
-uploads are commands that answer once the send queue has them. progress and
-failure arrive as events from one watcher on the sdk's global send queue
+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
 
 ## sessions and secrets
 
@@ -156,8 +176,12 @@ touching fifty rooms is one event
 
 ## verification
 
-`Confirm`, `Reject` and `Cancel`. reject tells the other side the emoji didn't
-match. the watcher and the action paths claim the terminal event, so a racing
+`Accept`, `Confirm`, `Reject` and `Cancel`. a request another device sends
+shows up as `Incoming` and waits for `Accept`, or `Cancel` to turn it down. one
+is only taken up while nothing else is in flight. reject tells the other side
+the emoji didn't match. the watchers subscribe before they look at the current
+state, since the sdk only reports what changes after you ask. the watcher and
+the action paths claim the terminal event, so a racing
 cancel can't turn `Done` into `Cancelled`
 
 ## public model

Modified file: docs/review-2026-09-24.md+28 additions−1 deletions@@ -249,7 +249,7 @@ Missing fields, all cheap SDK reads. Topic, canonical alias, joined and invited
 member counts, heroes, tombstone and successor, inviter for invites, a latest
 event preview, and space parents and children (Malice gap 3).
 
-### A6. No incoming verification. open
+### A6. No incoming verification. done, ported from malice's vendored copy
 
 Only outgoing `StartVerification` exists. A request started from another
 device, which is the normal "verify this new login" flow, is never observed.
@@ -393,3 +393,30 @@ avatar and encryption changes, which cover what Element shows inline, plus
 
 - `cargo fmt -- --check` clean.
 - `cargo test --all-targets` 49 passed, none failed.
+
+## follow-up review, 2026-09-25
+
+three parallel reviews of the reworked engine, each finding checked against the
+code before fixing. all fixed in the same change
+
+- close answered `Interrupted` when the actor had already gone, and anything
+  sent to an actor stopping after a leave was dropped. actors now say they're
+  leaving and hand back what they couldn't serve
+- closing a timeline released its media before stopping the watcher, so late
+  diffs re-registered it
+- thread receipts failed while the room was open and the thread wasn't
+- verification watchers subscribed after the state they needed, so a peer
+  starting sas left the flow waiting forever
+- a plain event could leave an encrypted file's ciphertext in the cache
+- downloads died at thirty seconds on the sdk client's timeout
+- `create_direct` made a second dm when asked twice before sync caught up
+- offline mode plus a server refusing sync bounced forever with no backoff
+- upload progress missed resumed uploads, failures could be lost to lag
+- `SendState::Failed { retriable }` meant the opposite of what a ui wants,
+  now `will_retry`, and `Queued` folded into `Sending`
+- room previews showed the `* edit` fallback, thumbnails claimed the original's
+  mimetype and ignored the asked size, v12 creators' power was dropped,
+  cleared names came back as empty strings
+- the example logged in afresh every run and broke its own store
+- adding the fixes tipped `Engine::run` past the compiler's query depth in every
+  consumer, so it now returns a boxed future

Modified file: examples/headless.rs+28 additions−13 deletions@@ -5,19 +5,28 @@ use macltok::{
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let session = Session::login_password(
-        PasswordLogin::new(
-            required("MATRIX_HOMESERVER")?,
-            required("MATRIX_USERNAME")?,
-            required("MATRIX_PASSWORD")?,
-        )
-        .with_device_display_name("macltok headless example"),
-        StoreConfig::new(
-            required("MATRIX_STORE_PATH")?,
-            required("MATRIX_STORE_PASSPHRASE")?,
-        ),
-    )
-    .await?;
+    let store_path = std::path::PathBuf::from(required("MATRIX_STORE_PATH")?);
+    let passphrase = required("MATRIX_STORE_PASSPHRASE")?;
+    // a real client keeps this somewhere safer, it holds the access token
+    let record_path = store_path.join("session.json");
+    let store = StoreConfig::new(&store_path, &passphrase);
+    let session = match std::fs::read(&record_path) {
+        Ok(saved) => Session::restore(serde_json::from_slice(&saved)?, store).await?,
+        Err(_) => {
+            let session = Session::login_password(
+                PasswordLogin::new(
+                    required("MATRIX_HOMESERVER")?,
+                    required("MATRIX_USERNAME")?,
+                    required("MATRIX_PASSWORD")?,
+                )
+                .with_device_display_name("macltok headless example"),
+                store,
+            )
+            .await?;
+            std::fs::write(&record_path, serde_json::to_vec(&session.record())?)?;
+            session
+        }
+    };
     let (engine, handle, mut events) = session.build(EngineConfig::default()).await?;
     let running = tokio::spawn(engine.run());
     let mut opened: Option<TimelineId> = None;
@@ -65,6 +74,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                         }
                         print_messages(&items);
                     }
+                    Event::SessionTokensChanged => {
+                        std::fs::write(&record_path, serde_json::to_vec(&handle.session_record()?)?)?;
+                    }
+                    Event::LoggedOut | Event::SessionAccessRejected { soft_logout: false } => {
+                        std::fs::remove_dir_all(&store_path)?;
+                    }
                     Event::Stopped => break,
                     _ => {}
                 }

Modified file: src/encryption.rs+87 additions−24 deletions@@ -68,6 +68,7 @@ impl VerificationController {
         action: VerificationAction,
     ) -> Result<(), Error> {
         match action {
+            VerificationAction::Accept => self.accept().await,
             VerificationAction::Confirm => {
                 // Retain a handle across the operation: the watcher clears
                 // `active` when it publishes a terminal event, but this clone
@@ -248,6 +249,9 @@ impl VerificationController {
                     })?
             }
         };
+        // subscribed before anything can wait on the event channel, or a quick peer's
+        // answer lands while we're blocked and nobody ever hears it
+        let changes = request.changes();
         self.active.lock().await.request = Some(request.clone());
         sender
             .send(Event::VerificationTargetChanged(target))
@@ -263,11 +267,80 @@ impl VerificationController {
             })?;
         let active = self.active.clone();
         self.task = Some(tokio::spawn(async move {
-            follow_request(request, active, sender).await;
+            follow_request(request, changes, active, sender).await;
         }));
         Ok(())
     }
 
+    /// another device asked us. only taken when nothing else is in flight, and not accepted
+    /// until the user says so with `VerificationAction::Accept`
+    pub(crate) async fn adopt_incoming(
+        &mut self,
+        sender: EventSender,
+        request: VerificationRequest,
+    ) {
+        if request.we_started() || request.is_done() || request.cancel_info().is_some() {
+            return;
+        }
+        if self.has_active_verification().await {
+            tracing::debug!("ignoring a verification request while another is in flight");
+            return;
+        }
+        self.stop_task().await;
+        let changes = request.changes();
+        self.active.lock().await.request = Some(request.clone());
+        let incoming = UiState::Incoming {
+            user_id: request.other_user_id().to_owned(),
+            device_id: match request.state() {
+                VerificationRequestState::Requested {
+                    other_device_data, ..
+                }
+                | VerificationRequestState::Ready {
+                    other_device_data, ..
+                } => Some(other_device_data.device_id().to_owned()),
+                _ => None,
+            },
+        };
+        let target = VerificationTarget::User(request.other_user_id().to_owned());
+        for event in [
+            Event::VerificationTargetChanged(target),
+            Event::VerificationChanged(incoming),
+        ] {
+            if sender.send(event).await.is_err() {
+                return;
+            }
+        }
+        let active = self.active.clone();
+        self.task = Some(tokio::spawn(async move {
+            follow_request(request, changes, active, sender).await;
+        }));
+    }
+
+    async fn accept(&self) -> Result<(), Error> {
+        let inactive = |message: &str| Error::new(ErrorKind::VerificationInactive, message);
+        let (request, sas) = {
+            let active = self.active.lock().await;
+            (active.request.clone(), active.sas.is_some())
+        };
+        let request = request.ok_or_else(|| inactive("no verification request is pending"))?;
+        if request.we_started() {
+            return Err(inactive("nobody asked us, we asked them"));
+        }
+        if request.is_done() {
+            return Err(inactive("verification is already complete"));
+        }
+        if let Some(info) = request.cancel_info() {
+            return Err(inactive(info.reason()));
+        }
+        if sas {
+            return Err(inactive("verification has already started"));
+        }
+        request
+            .accept_with_methods(vec![VerificationMethod::SasV1])
+            .await
+            .map_err(|source| Error::from_sdk("accepting verification request failed", source))
+    }
+
     async fn confirm(&self) -> Result<(), Error> {
         let sas = self.active.lock().await.sas.clone().ok_or_else(|| {
             Error::new(
@@ -560,33 +633,20 @@ async fn request_identity_verification(
         .map_err(|source| Error::from_verification_request("starting verification failed", source))
 }
 
+/// `changes` is subscribed by the caller before the request could move, and the current state
+/// is handled first, since the stream only reports what happens after it was taken
 async fn follow_request(
     request: VerificationRequest,
+    changes: impl futures_util::Stream<Item = VerificationRequestState> + Unpin,
     active: Arc<Mutex<VerificationSession>>,
     sender: EventSender,
 ) {
-    let mut changes = request.changes();
-    while let Some(state) = changes.next().await {
+    let mut states = futures_util::stream::iter([request.state()]).chain(changes);
+    while let Some(state) = states.next().await {
         match state {
-            VerificationRequestState::Created { .. } => {}
-            VerificationRequestState::Requested { .. } => {
-                if let Err(error) = request
-                    .accept_with_methods(vec![VerificationMethod::SasV1])
-                    .await
-                {
-                    tracing::warn!(error = %error, "accepting verification request failed");
-                    // Claim the session so this terminal outcome is emitted exactly
-                    // once; a lost claim means an action path already owns it.
-                    report_watcher_terminal(
-                        &active,
-                        &sender,
-                        UiState::Failed(VerificationFailure::RequestAcceptance(error.to_string())),
-                        "request acceptance failure",
-                    )
-                    .await;
-                    return;
-                }
-            }
+            // created is ours waiting on them, requested is theirs waiting on `Accept`
+            VerificationRequestState::Created { .. }
+            | VerificationRequestState::Requested { .. } => {}
             VerificationRequestState::Ready { .. } => match request.start_sas().await {
                 Ok(Some(sas)) => {
                     follow_sas(sas, active.clone(), sender.clone()).await;
@@ -664,9 +724,12 @@ async fn follow_sas(
     active: Arc<Mutex<VerificationSession>>,
     sender: EventSender,
 ) {
+    // the sdk may already have moved the flow along before we got here, a peer starting
+    // sas is the usual case, so take the stream first and the current state after
+    let changes = sas.changes();
     active.lock().await.sas = Some(sas.clone());
-    let mut changes = sas.changes();
-    while let Some(state) = changes.next().await {
+    let mut states = futures_util::stream::iter([sas.state()]).chain(changes);
+    while let Some(state) = states.next().await {
         let update = match state {
             SasState::Created { .. } | SasState::Accepted { .. } => continue,
             SasState::Started { .. } => {

Modified file: src/engine/commands.rs+13 additions−3 deletions@@ -1,6 +1,7 @@
 use std::{future::Future, panic::AssertUnwindSafe};
 
 use futures_util::FutureExt;
+use matrix_sdk::encryption::verification::VerificationRequest;
 use matrix_sdk::{
     Client,
     notification_settings::{IsEncrypted, IsOneToOne, RoomNotificationMode as SdkMode},
@@ -48,11 +49,20 @@ pub(super) async fn run_encryption(
     client: Client,
     events: EventSender,
     mut jobs: mpsc::Receiver<Job<EncryptionCommand>>,
+    mut incoming: mpsc::UnboundedReceiver<VerificationRequest>,
 ) {
     let mut verification = VerificationController::default();
-    while let Some(Job { command, reply }) = jobs.recv().await {
-        let work = encryption(&client, &events, &mut verification, command);
-        reply.done(guarded("encryption command", work).await);
+    loop {
+        tokio::select! {
+            job = jobs.recv() => {
+                let Some(Job { command, reply }) = job else { return };
+                let work = encryption(&client, &events, &mut verification, command);
+                reply.done(guarded("encryption command", work).await);
+            }
+            Some(request) = incoming.recv() => {
+                verification.adopt_incoming(events.clone(), request).await;
+            }
+        }
     }
 }
 

Modified file: src/engine/media.rs+93 additions−32 deletions@@ -4,6 +4,7 @@ use std::{
     io::Read as _,
     panic::AssertUnwindSafe,
     sync::Arc,
+    time::Duration,
 };
 
 use futures_util::{
@@ -36,8 +37,10 @@ use crate::{
     upload::prepare_upload,
 };
 
-/// the uploader's thumbnail is used when there is one, this is only what we ask the server for
-const TIMELINE_THUMBNAIL: u32 = 800;
+/// a stalled download gives up after this long without a byte
+const DOWNLOAD_IDLE: Duration = Duration::from_secs(60);
+/// the sdk client gives up on any request after thirty seconds, far too soon for a video
+const DOWNLOAD_TOTAL: Duration = Duration::from_secs(60 * 60);
 
 type FlightFuture = BoxFuture<'static, Result<Media, Error>>;
 type Flight = Shared<FlightFuture>;
@@ -99,7 +102,7 @@ impl MediaService {
                 media.full.mimetype,
                 media.full.declared_size,
             ),
-            MediaSize::Thumbnail { .. } => match media.thumbnail {
+            MediaSize::Thumbnail { width, height } => match media.thumbnail {
                 Some(thumbnail) => (
                     thumbnail.source,
                     MediaFormat::File,
@@ -110,12 +113,13 @@ impl MediaService {
                 None if matches!(media.kind, MediaKind::Image | MediaKind::Video)
                     && matches!(media.full.source, MediaSource::Plain(_)) =>
                 {
-                    (
-                        media.full.source,
-                        thumbnail_format(TIMELINE_THUMBNAIL, TIMELINE_THUMBNAIL),
-                        media.full.mimetype,
-                        None,
-                    )
+                    // the server picks the thumbnail's type, so the original's would be a lie
+                    let format = if is_local(&media.full.source) {
+                        MediaFormat::File
+                    } else {
+                        thumbnail_format(width, height)
+                    };
+                    (media.full.source, format, None, None)
                 }
                 None => {
                     return Err(Error::new(
@@ -128,7 +132,7 @@ impl MediaService {
         if declared_size.is_some_and(|size| size > self.max_bytes as u64) {
             return Err(too_large("media exceeds the configured byte limit"));
         }
-        self.fetch(MediaRequestParameters { source, format }, mimetype)
+        self.fetch_or_file(MediaRequestParameters { source, format }, mimetype)
             .await
     }
 
@@ -140,7 +144,7 @@ impl MediaService {
             MediaSize::Full => MediaFormat::File,
             MediaSize::Thumbnail { width, height } => thumbnail_format(width, height),
         };
-        self.fetch(
+        self.fetch_or_file(
             MediaRequestParameters {
                 source: MediaSource::Plain(uri.to_owned()),
                 format,
@@ -150,6 +154,28 @@ impl MediaService {
         .await
     }
 
+    /// plenty of servers won't thumbnail everything, and the file itself is the next best
+    /// thing. the byte cap still holds either way
+    async fn fetch_or_file(
+        &self,
+        request: MediaRequestParameters,
+        mimetype: Option<String>,
+    ) -> Result<Media, Error> {
+        if !matches!(request.format, MediaFormat::Thumbnail(_)) {
+            return self.fetch(request, mimetype).await;
+        }
+        match self.fetch(request.clone(), mimetype.clone()).await {
+            Err(error) if error.kind == ErrorKind::MediaUnavailable => {
+                let file = MediaRequestParameters {
+                    source: request.source,
+                    format: MediaFormat::File,
+                };
+                self.fetch(file, mimetype).await
+            }
+            result => result,
+        }
+    }
+
     async fn fetch(
         &self,
         request: MediaRequestParameters,
@@ -191,10 +217,20 @@ fn thumbnail_format(width: u32, height: u32) -> MediaFormat {
     MediaFormat::Thumbnail(MediaThumbnailSettings::new(width.into(), height.into()))
 }
 
+fn is_local(source: &MediaSource) -> bool {
+    let uri = match source {
+        MediaSource::Plain(uri) => uri,
+        MediaSource::Encrypted(file) => &file.url,
+    };
+    uri.server_name()
+        .is_ok_and(|server| server.as_str() == "send-queue.localhost")
+}
+
+/// plain and encrypted sources share a cache key in the sdk, but must never share a flight
 fn flight_key(request: &MediaRequestParameters) -> String {
     let uri = match &request.source {
-        MediaSource::Plain(uri) => uri.as_str(),
-        MediaSource::Encrypted(file) => file.url.as_str(),
+        MediaSource::Plain(uri) => format!("plain|{uri}"),
+        MediaSource::Encrypted(file) => format!("encrypted|{}", file.url),
     };
     match &request.format {
         MediaFormat::File => uri.to_owned(),
@@ -222,21 +258,42 @@ async fn fetch(
             }
             return Err(too_large("cached media exceeds the configured byte limit"));
         }
+        // plain and encrypted share a cache key, so a plain fetch of an encrypted file's uri
+        // leaves ciphertext here. its hash gives it away
+        if is_ciphertext(&cached, &request.source) {
+            let bytes = maybe_decrypt(cached, &request, max_bytes)?;
+            cache(&client, &request, &bytes).await;
+            return Ok(media(bytes, mimetype));
+        }
         return Ok(media(cached, mimetype));
     }
     let (bytes, served_mimetype) = download_bounded(&client, &request, max_bytes).await?;
     // the sdk caches plaintext under the encrypted key, so we do too or nobody can read anything
     let bytes = maybe_decrypt(bytes, &request, max_bytes)?;
+    cache(&client, &request, &bytes).await;
+    Ok(media(bytes, mimetype.or(served_mimetype)))
+}
+
+async fn cache(client: &Client, request: &MediaRequestParameters, bytes: &[u8]) {
     if let Ok(store) = client.media_store().lock().await {
         let _ = store
             .add_media_content(
-                &request,
-                bytes.clone(),
+                request,
+                bytes.to_vec(),
                 matrix_sdk::media::store::IgnoreMediaRetentionPolicy::No,
             )
             .await;
     }
-    Ok(media(bytes, mimetype.or(served_mimetype)))
+}
+
+fn is_ciphertext(bytes: &[u8], source: &MediaSource) -> bool {
+    use sha2::{Digest, Sha256};
+    let MediaSource::Encrypted(file) = source else {
+        return false;
+    };
+    file.hashes
+        .get(&ruma::events::room::EncryptedFileHashAlgorithm::Sha256)
+        .is_some_and(|hash| hash.as_bytes() == Sha256::digest(bytes).as_slice())
 }
 
 fn media(bytes: Vec<u8>, mimetype: Option<String>) -> Media {
@@ -325,20 +382,13 @@ fn maybe_decrypt(
     }
 }
 
-/// the sdk's own download has no size cap and no timeout, so we do it by hand
+/// the sdk's own download has no size cap, so we do it by hand
 async fn download_bounded(
     client: &Client,
     request: &MediaRequestParameters,
     max_bytes: usize,
 ) -> Result<(Vec<u8>, Option<String>), Error> {
-    let uri = match &request.source {
-        MediaSource::Plain(uri) => uri,
-        MediaSource::Encrypted(file) => &file.url,
-    };
-    if uri
-        .server_name()
-        .is_ok_and(|server| server.as_str() == "send-queue.localhost")
-    {
+    if is_local(&request.source) {
         return Err(Error::new(
             ErrorKind::MediaUnavailable,
             "local media is not in the media cache",
@@ -350,7 +400,10 @@ async fn download_bounded(
     let response = loop {
         let http_request = media_http_request(client, request).await?;
         let (parts, _) = http_request.into_parts();
-        let mut builder = client.http_client().get(parts.uri.to_string());
+        let mut builder = client
+            .http_client()
+            .get(parts.uri.to_string())
+            .timeout(DOWNLOAD_TOTAL);
         for (name, value) in &parts.headers {
             builder = builder.header(name.as_str(), value.as_bytes());
         }
@@ -393,7 +446,11 @@ async fn download_bounded(
         .map(str::to_owned);
     let mut stream = response.bytes_stream();
     let mut bytes = Vec::new();
-    while let Some(chunk) = stream.next().await {
+    loop {
+        let chunk = tokio::time::timeout(DOWNLOAD_IDLE, stream.next())
+            .await
+            .map_err(|_| Error::new(ErrorKind::Network, "media download stalled"))?;
+        let Some(chunk) = chunk else { break };
         let chunk = chunk.map_err(network_failed)?;
         if chunk.len() > max_bytes.saturating_sub(bytes.len()) {
             return Err(too_large(
@@ -501,11 +558,15 @@ pub(crate) async fn queue_upload(
         .await
         .map(drop)
         .map_err(|source| {
-            Error::with_source(
-                ErrorKind::MediaUnavailable,
-                "media upload could not be queued",
-                source,
-            )
+            use matrix_sdk::send_queue::RoomSendQueueError;
+            let kind = match &source {
+                RoomSendQueueError::RoomNotJoined | RoomSendQueueError::RoomDisappeared => {
+                    ErrorKind::RoomUnknown
+                }
+                RoomSendQueueError::StorageError(_) => ErrorKind::Store,
+                _ => ErrorKind::MediaUnavailable,
+            };
+            Error::with_source(kind, "media upload could not be queued", source)
         })
 }
 

Modified file: src/engine/mod.rs+163 additions−58 deletions@@ -14,7 +14,7 @@ use std::{
 use futures_util::future::BoxFuture;
 use matrix_sdk::{Client, SessionChange};
 use matrix_sdk_ui::sync_service::SyncService;
-use ruma::{OwnedRoomId, RoomId};
+use ruma::OwnedRoomId;
 use thiserror::Error as ThisError;
 use tokio::{
     sync::{Notify, broadcast, mpsc, watch},
@@ -26,7 +26,7 @@ pub(crate) use media::MediaService;
 use self::{
     commands::Job,
     mutations::{Mutation, MutationJob},
-    room::{ActorRequest, Exited, RoomActor},
+    room::{ActorNote, ActorRequest, RoomActor},
     timeline::{SessionContext, Supervisor},
 };
 use crate::{
@@ -208,6 +208,10 @@ struct Queues {
     session_changes: broadcast::Receiver<SessionChange>,
 }
 
+/// requests other devices send us, handed from the sdk's event handler to the encryption worker
+type IncomingVerifications =
+    mpsc::UnboundedSender<matrix_sdk::encryption::verification::VerificationRequest>;
+
 pub struct Engine {
     client: Client,
     sync: Arc<SyncService>,
@@ -217,14 +221,18 @@ pub struct Engine {
     tasks: TaskRegistry,
     rooms: HashMap<OwnedRoomId, RoomSlot>,
     next_generation: u64,
-    exits: mpsc::UnboundedReceiver<Exited>,
-    exit_sender: mpsc::UnboundedSender<Exited>,
+    notes: mpsc::UnboundedReceiver<ActorNote>,
+    note_sender: mpsc::UnboundedSender<ActorNote>,
+    /// the rooms sliding sync should follow, applied off the run loop since it touches the store
+    subscriptions: watch::Sender<Vec<OwnedRoomId>>,
     mutations: mpsc::Sender<MutationJob>,
     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, (AbortHandle, watch::Receiver<bool>)>,
+    /// 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: Arc<Notify>,
 }
 
@@ -264,7 +272,7 @@ impl Engine {
             config.event_capacity,
         );
         client.send_queue().enable_upload_progress(true);
-        let (exit_sender, exits) = mpsc::unbounded_channel();
+        let (note_sender, notes) = mpsc::unbounded_channel();
         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);
@@ -280,8 +288,9 @@ impl Engine {
             },
             rooms: HashMap::new(),
             next_generation: 0,
-            exits,
-            exit_sender,
+            notes,
+            note_sender,
+            subscriptions: watch::Sender::new(Vec::new()),
             mutations,
             notifications,
             encryption,
@@ -292,12 +301,19 @@ impl Engine {
                 session_changes,
             }),
             uploads: HashMap::new(),
+            verification_handler: None,
             stop: Arc::new(Notify::new()),
         };
         Ok((engine, handle, events))
     }
 
-    pub async fn run(mut self) -> Result<(), EngineError> {
+    /// boxed because the engine's future is deep enough to blow the compiler's query depth
+    /// in every crate that spawns it, which isn't something a client should have to fix
+    pub fn run(self) -> BoxFuture<'static, Result<(), EngineError>> {
+        Box::pin(self.run_inner())
+    }
+
+    async fn run_inner(mut self) -> Result<(), EngineError> {
         let events = self.port.event_sender();
         let user_id = self
             .client
@@ -317,6 +333,7 @@ impl Engine {
             }
         }
         self.start_tasks();
+        self.resume_upload_forwarders().await;
         self.sync.start().await;
 
         let mut failure = None;
@@ -329,13 +346,6 @@ impl Engine {
                     break;
                 }
                 () = self.stop.notified() => break,
-                request = self.port.recv() => {
-                    let Some(request) = request else { break };
-                    if let Some(reply) = self.dispatch(request).await {
-                        shutdown_reply = Some(reply);
-                        break;
-                    }
-                }
                 Some((context, result)) = self.tasks.join_next(), if !self.tasks.tasks.is_empty() => {
                     let Err(error) = result else { continue };
                     if !error.is_panic() {
@@ -348,7 +358,14 @@ impl Engine {
                     failure = Some(EngineError::TaskPanicked { task: context.label() });
                     break;
                 }
-                Some(exited) = self.exits.recv() => self.room_exited(exited).await,
+                Some(note) = self.notes.recv() => self.actor_note(note).await,
+                request = self.port.recv() => {
+                    let Some(request) = request else { break };
+                    if let Some(reply) = self.dispatch(request).await {
+                        shutdown_reply = Some(reply);
+                        break;
+                    }
+                }
             }
         }
 
@@ -360,14 +377,37 @@ impl Engine {
         if let Some(reply) = shutdown_reply {
             reply.done(Ok(()));
         }
+        // everything else still queued answers `Interrupted` as it drops, but a shutdown did happen
+        while let Some(Request { command, reply }) = self.port.try_recv() {
+            if matches!(command, Command::Lifecycle(LifecycleCommand::Shutdown)) {
+                reply.done(Ok(()));
+            }
+        }
         failure.map_or(Ok(()), Err)
     }
 
     fn start_tasks(&mut self) {
         let queues = self.queues.take().expect("engine tasks start once");
+        let (incoming, incoming_verifications) = mpsc::unbounded_channel();
+        self.verification_handler = Some(self.verification_handler(incoming));
         let client = &self.client;
         let events = || self.port.event_sender();
-        let watchers: [(&'static str, BoxFuture<'static, ()>); 9] = [
+        let watchers: [(&'static str, BoxFuture<'static, ()>); 11] = [
+            (
+                "upload failure watcher",
+                Box::pin(watchers::upload_failures(
+                    client.clone(),
+                    client.send_queue().subscribe(),
+                    events(),
+                )),
+            ),
+            (
+                "room subscription watcher",
+                Box::pin(watchers::room_subscriptions(
+                    self.sync.clone(),
+                    self.subscriptions.subscribe(),
+                )),
+            ),
             (
                 "session change watcher",
                 Box::pin(watchers::session(
@@ -427,6 +467,7 @@ impl Engine {
                     client.clone(),
                     events(),
                     queues.encryption,
+                    incoming_verifications,
                 )),
             ),
         ];
@@ -435,6 +476,30 @@ impl Engine {
         }
     }
 
+    fn verification_handler(
+        &self,
+        incoming: IncomingVerifications,
+    ) -> matrix_sdk::event_handler::EventHandlerHandle {
+        use matrix_sdk::ruma::events::key::verification::request::ToDeviceKeyVerificationRequestEvent;
+        self.client.add_event_handler(
+            move |event: ToDeviceKeyVerificationRequestEvent, client: Client| {
+                let incoming = incoming.clone();
+                async move {
+                    let request = client
+                        .encryption()
+                        .get_verification_request(
+                            &event.sender,
+                            event.content.transaction_id.as_str(),
+                        )
+                        .await;
+                    if let Some(request) = request {
+                        let _ = incoming.send(request);
+                    }
+                }
+            },
+        )
+    }
+
     /// hands back the shutdown request's responder, which is answered once we've stopped
     async fn dispatch(&mut self, Request { command, reply }: Request) -> Option<Responder> {
         let client = self.client.clone();
@@ -552,21 +617,7 @@ impl Engine {
         let Some(room) = self.client.get_room(upload.timeline_id.room_id()) else {
             return reply.done(Err(Error::room_unknown()));
         };
-        let room_id = room.room_id().to_owned();
-        self.uploads
-            .retain(|_, (forwarder, _)| !forwarder.is_finished());
-        let mut ready = match self.uploads.get(&room_id) {
-            Some((_, ready)) => ready.clone(),
-            None => {
-                let (subscribed, ready) = watch::channel(false);
-                let forwarder = self.tasks.spawn(
-                    TaskContext::Command("upload watcher"),
-                    watchers::room_uploads(room.clone(), self.port.event_sender(), subscribed),
-                );
-                self.uploads.insert(room_id, (forwarder, ready.clone()));
-                ready
-            }
-        };
+        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 {
@@ -578,6 +629,34 @@ impl Engine {
         });
     }
 
+    fn upload_forwarder(&mut self, room: &matrix_sdk::Room) -> watch::Receiver<bool> {
+        self.uploads
+            .retain(|_, (forwarder, _)| !forwarder.is_finished());
+        if let Some((_, ready)) = self.uploads.get(room.room_id()) {
+            return ready.clone();
+        }
+        let (subscribed, ready) = watch::channel(false);
+        let forwarder = self.tasks.spawn(
+            TaskContext::Command("upload watcher"),
+            watchers::room_uploads(room.clone(), self.port.event_sender(), subscribed),
+        );
+        self.uploads
+            .insert(room.room_id().to_owned(), (forwarder, 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()),
@@ -624,8 +703,9 @@ impl Engine {
             Some(RoomSlot::Active { generation, sender }) => (*generation, sender.clone()),
         };
         match sender.try_send(ActorRequest::Close(reply)) {
-            // a closed channel means the actor already left on its own, which is closed enough
-            Ok(()) | Err(mpsc::error::TrySendError::Closed(_)) => {}
+            Ok(()) => {}
+            // the actor already left on its own, which is closed enough
+            Err(mpsc::error::TrySendError::Closed(request)) => request.into_reply().done(Ok(())),
             Err(mpsc::error::TrySendError::Full(request)) => {
                 return request.into_reply().done(Err(Error::new(
                     ErrorKind::RoomBusy,
@@ -640,7 +720,7 @@ impl Engine {
                 pending: VecDeque::new(),
             },
         );
-        self.update_room_subscriptions().await;
+        self.update_room_subscriptions();
     }
 
     async fn start_actor(&mut self, room_id: OwnedRoomId, reply: Responder) {
@@ -656,7 +736,7 @@ impl Engine {
             generation,
             room,
             requests,
-            exits: self.exit_sender.clone(),
+            notes: self.note_sender.clone(),
             context: SessionContext {
                 events: self.port.event_sender(),
                 media: self.media.clone(),
@@ -676,16 +756,21 @@ impl Engine {
             },
             actor.run(),
         );
-        self.update_room_subscriptions().await;
+        self.update_room_subscriptions();
     }
 
-    async fn room_exited(
-        &mut self,
-        Exited {
-            room_id,
-            generation,
-        }: Exited,
-    ) {
+    async fn actor_note(&mut self, note: ActorNote) {
+        let (room_id, generation) = match &note {
+            ActorNote::Leaving {
+                room_id,
+                generation,
+            }
+            | ActorNote::Exited {
+                room_id,
+                generation,
+                ..
+            } => (room_id.clone(), *generation),
+        };
         if self
             .rooms
             .get(&room_id)
@@ -693,31 +778,51 @@ impl Engine {
         {
             return;
         }
-        if let Some(RoomSlot::Closing { pending, .. }) = self.rooms.remove(&room_id) {
-            for deferred in pending {
-                match deferred {
-                    Deferred::Open(reply) => self.open_room(room_id.clone(), reply).await,
-                    Deferred::Close(reply) => self.close_room(room_id.clone(), reply).await,
-                }
+        let ActorNote::Exited { leftovers, .. } = note else {
+            if let Some(slot @ RoomSlot::Active { .. }) = self.rooms.get_mut(&room_id) {
+                *slot = RoomSlot::Closing {
+                    generation,
+                    pending: VecDeque::new(),
+                };
+                self.update_room_subscriptions();
+            }
+            return;
+        };
+        let pending = match self.rooms.remove(&room_id) {
+            Some(RoomSlot::Closing { pending, .. }) => pending,
+            _ => VecDeque::new(),
+        };
+        // leftovers were sent before anything deferred, so they go first
+        for request in leftovers {
+            match request {
+                ActorRequest::Open(reply) => self.open_room(room_id.clone(), reply).await,
+                ActorRequest::Close(reply) => reply.done(Ok(())),
+                ActorRequest::Execute(command, reply) => self.route_timeline(command, reply),
             }
         }
-        self.update_room_subscriptions().await;
+        for deferred in pending {
+            match deferred {
+                Deferred::Open(reply) => self.open_room(room_id.clone(), reply).await,
+                Deferred::Close(reply) => self.close_room(room_id.clone(), reply).await,
+            }
+        }
+        self.update_room_subscriptions();
     }
 
-    async fn update_room_subscriptions(&self) {
+    fn update_room_subscriptions(&self) {
         let room_ids = self
             .rooms
             .iter()
             .filter(|(_, slot)| matches!(slot, RoomSlot::Active { .. }))
-            .map(|(room_id, _)| room_id.as_ref())
-            .collect::<Vec<&RoomId>>();
-        self.sync
-            .room_list_service()
-            .subscribe_to_rooms(&room_ids)
-            .await;
+            .map(|(room_id, _)| room_id.clone())
+            .collect();
+        self.subscriptions.send_replace(room_ids);
     }
 
     async fn shutdown(&mut self) {
+        if let Some(handler) = self.verification_handler.take() {
+            self.client.remove_event_handler(handler);
+        }
         self.port.close();
         self.tasks.shutdown().await;
         self.rooms.clear();

Modified file: src/engine/mutations.rs+55 additions−14 deletions@@ -1,4 +1,7 @@
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::{
+    collections::{HashMap, HashSet, VecDeque},
+    sync::Arc,
+};
 
 use matrix_sdk::{
     Client, Room,
@@ -50,6 +53,7 @@ pub(super) async fn run_scheduler(
     mut jobs: mpsc::Receiver<MutationJob>,
     queue_limit: usize,
 ) {
+    let fresh_dms = FreshDms::default();
     let mut waiting = HashMap::<Key, VecDeque<MutationJob>>::new();
     let mut running = HashSet::<Key>::new();
     let mut tasks = JoinSet::<Key>::new();
@@ -69,7 +73,7 @@ pub(super) async fn run_scheduler(
                 match next {
                     Some(job) => {
                         queued -= 1;
-                        spawn(&mut tasks, &client, key, job);
+                        spawn(&mut tasks, &client, &fresh_dms, key, job);
                     }
                     None => {
                         running.remove(&key);
@@ -81,12 +85,12 @@ pub(super) async fn run_scheduler(
                     receiving = false;
                     continue;
                 };
-                let key = key(&job.command).unwrap_or_else(|| {
+                let key = key(&job.command, client.user_id().map(ruma::UserId::server_name)).unwrap_or_else(|| {
                     independent += 1;
                     Key::Independent(independent)
                 });
                 if running.insert(key.clone()) {
-                    spawn(&mut tasks, &client, key, job);
+                    spawn(&mut tasks, &client, &fresh_dms, key, job);
                 } else if queued < queue_limit {
                     waiting.entry(key).or_default().push_back(job);
                     queued += 1;
@@ -101,7 +105,8 @@ pub(super) async fn run_scheduler(
     }
 }
 
-fn key(mutation: &Mutation) -> Option<Key> {
+/// `server` is ours, since a new alias lives there
+fn key(mutation: &Mutation, server: Option<&ruma::ServerName>) -> Option<Key> {
     let room_id = match mutation {
         Mutation::GlobalAccountData { event_type, .. } => {
             return Some(Key::AccountData(event_type.to_string()));
@@ -115,6 +120,13 @@ fn key(mutation: &Mutation) -> Option<Key> {
         Mutation::Room(RoomCommand::CreateDirect(user_id)) => {
             return Some(Key::Direct(user_id.clone()));
         }
+        Mutation::Room(RoomCommand::Create(options)) => {
+            return match (&options.alias_localpart, options.invite.as_slice()) {
+                (Some(localpart), _) => Some(Key::Alias(format!("#{localpart}:{}", server?))),
+                (None, [user_id]) if options.direct => Some(Key::Direct(user_id.clone())),
+                _ => None,
+            };
+        }
         Mutation::Room(
             RoomCommand::Leave(room_id)
             | RoomCommand::Forget(room_id)
@@ -133,16 +145,31 @@ fn key(mutation: &Mutation) -> Option<Key> {
     Some(Key::Room(room_id.clone()))
 }
 
-fn spawn(tasks: &mut JoinSet<Key>, client: &Client, key: Key, job: MutationJob) {
+/// dms made here that the next sync hasn't told the sdk about yet, since `get_dm_room`
+/// only knows `m.direct` once it comes back round
+type FreshDms = Arc<std::sync::Mutex<HashMap<OwnedUserId, OwnedRoomId>>>;
+
+fn spawn(
+    tasks: &mut JoinSet<Key>,
+    client: &Client,
+    fresh_dms: &FreshDms,
+    key: Key,
+    job: MutationJob,
+) {
     let client = client.clone();
+    let fresh_dms = fresh_dms.clone();
     tasks.spawn(async move {
         let Job { command, reply } = job;
-        reply.send(guarded("room command", execute(&client, command)).await);
+        reply.send(guarded("room command", execute(&client, &fresh_dms, command)).await);
         key
     });
 }
 
-async fn execute(client: &Client, mutation: Mutation) -> Result<CommandOutput, Error> {
+async fn execute(
+    client: &Client,
+    fresh_dms: &FreshDms,
+    mutation: Mutation,
+) -> Result<CommandOutput, Error> {
     let command = match mutation {
         Mutation::GlobalAccountData {
             event_type,
@@ -246,14 +273,28 @@ async fn execute(client: &Client, mutation: Mutation) -> Result<CommandOutput, E
             .await
             .map(joined)
             .map_err(sdk("creating room failed")),
-        RoomCommand::CreateDirect(user_id) => match client.get_dm_room(&user_id) {
-            Some(existing) => Ok(joined(existing)),
-            None => client
+        RoomCommand::CreateDirect(user_id) => {
+            if let Some(existing) = client.get_dm_room(&user_id) {
+                return Ok(joined(existing));
+            }
+            let fresh = fresh_dms
+                .lock()
+                .expect("nothing panics holding this")
+                .get(&user_id)
+                .cloned();
+            if let Some(room_id) = fresh.filter(|room_id| client.get_room(room_id).is_some()) {
+                return Ok(CommandOutput::Room(room_id));
+            }
+            let room = client
                 .create_dm(&user_id)
                 .await
-                .map(joined)
-                .map_err(sdk("creating direct room failed")),
-        },
+                .map_err(sdk("creating direct room failed"))?;
+            fresh_dms
+                .lock()
+                .expect("nothing panics holding this")
+                .insert(user_id, room.room_id().to_owned());
+            Ok(joined(room))
+        }
         _ => Err(Error::new(ErrorKind::Other, "room command was misrouted")),
     }
 }

Modified file: src/engine/room.rs+52 additions−13 deletions@@ -8,7 +8,7 @@ use tokio::{
     task::{AbortHandle, JoinSet},
 };
 
-use super::timeline::{SessionContext, TimelineSession};
+use super::timeline::{SessionContext, TaskGuard, TimelineSession};
 use crate::{
     Error, ErrorKind, Event, TimelineCommand, TimelineId, TypingState, transport::Responder,
 };
@@ -28,9 +28,18 @@ impl ActorRequest {
     }
 }
 
-pub(super) struct Exited {
-    pub(super) room_id: OwnedRoomId,
-    pub(super) generation: u64,
+pub(super) enum ActorNote {
+    /// stopping on its own, after a leave or a kick. anything sent its way now comes back in `Exited`
+    Leaving {
+        room_id: OwnedRoomId,
+        generation: u64,
+    },
+    Exited {
+        room_id: OwnedRoomId,
+        generation: u64,
+        /// requests that arrived too late to be served, for the engine to answer or replay
+        leftovers: Vec<ActorRequest>,
+    },
 }
 
 pub(super) struct RoomActor {
@@ -38,7 +47,7 @@ pub(super) struct RoomActor {
     pub(super) generation: u64,
     pub(super) room: Room,
     pub(super) requests: mpsc::Receiver<ActorRequest>,
-    pub(super) exits: mpsc::UnboundedSender<Exited>,
+    pub(super) notes: mpsc::UnboundedSender<ActorNote>,
     pub(super) context: SessionContext,
     pub(super) panics: mpsc::UnboundedReceiver<&'static str>,
     pub(super) opening: Vec<Responder>,
@@ -46,13 +55,17 @@ pub(super) struct RoomActor {
 
 impl RoomActor {
     pub(super) async fn run(mut self) {
-        let exits = self.exits.clone();
-        let exited = Exited {
+        self.serve().await;
+        self.requests.close();
+        let mut leftovers = Vec::new();
+        while let Ok(request) = self.requests.try_recv() {
+            leftovers.push(request);
+        }
+        let _ = self.notes.send(ActorNote::Exited {
             room_id: self.room_id.clone(),
             generation: self.generation,
-        };
-        self.serve().await;
-        let _ = exits.send(exited);
+            leftovers,
+        });
     }
 
     async fn serve(&mut self) {
@@ -80,21 +93,28 @@ impl RoomActor {
             reply.done(Ok(()));
         }
         let (typing_guard, typing) = self.room.subscribe_to_typing_notifications();
-        let typing_watcher = self.context.supervisor.spawn(
+        let typing_watcher = TaskGuard(self.context.supervisor.spawn(
             "typing watcher",
             watch_typing(self.room_id.clone(), typing, self.context.events.clone()),
-        );
+        ));
         let mut timelines = Timelines {
             room: room_timeline,
             threads: Threads::new(self.requests.max_capacity()),
             work: JoinSet::new(),
         };
         let close = self.serve_open(&mut timelines).await;
+        if close.is_none() {
+            self.requests.close();
+            let _ = self.notes.send(ActorNote::Leaving {
+                room_id: self.room_id.clone(),
+                generation: self.generation,
+            });
+        }
 
         // let redactions and receipts that already left finish rather than lie about them
         while timelines.work.join_next().await.is_some() {}
         drop(typing_guard);
-        typing_watcher.abort();
+        drop(typing_watcher);
         timelines.threads.close().await;
         timelines.room.close().await;
         if let Some(reply) = close {
@@ -105,6 +125,10 @@ impl RoomActor {
     /// returns the close request's responder, if a close is why we stopped
     async fn serve_open(&mut self, timelines: &mut Timelines) -> Option<Responder> {
         let mut room_info = self.room.subscribe_info();
+        // a leave that landed while the timeline was building won't come round again
+        if matches!(self.room.state(), RoomState::Left | RoomState::Banned) {
+            return None;
+        }
         loop {
             tokio::select! {
                 Some(info) = room_info.next() => {
@@ -208,6 +232,21 @@ impl Timelines {
                 let Some(root) = root else {
                     return self.room.execute(command, reply, &mut self.work).await;
                 };
+                if let TimelineCommand::MarkRead {
+                    timeline_id,
+                    event_id,
+                } = &command
+                    && !self.threads.open.contains_key(&root)
+                {
+                    let client = room.client();
+                    let (timeline_id, event_id) = (timeline_id.clone(), event_id.clone());
+                    self.work.spawn(async move {
+                        reply.done(
+                            super::commands::mark_read(&client, &timeline_id, event_id).await,
+                        );
+                    });
+                    return;
+                }
                 match self.threads.route(root, command, reply) {
                     Routed::Open(session, TimelineCommand::Resync { .. }, reply) => {
                         reply.done(session.resync().await);

Modified file: src/engine/timeline.rs+11 additions−1 deletions@@ -55,6 +55,15 @@ impl Supervisor {
     }
 }
 
+/// aborts its task when dropped, so an actor that dies mid-flight takes its helpers with it
+pub(super) struct TaskGuard(pub(super) JoinHandle<()>);
+
+impl Drop for TaskGuard {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}
+
 /// what a timeline session needs from the room that owns it
 #[derive(Clone)]
 pub(super) struct SessionContext {
@@ -145,11 +154,12 @@ impl TimelineSession {
 
     /// pending paginations answer `Interrupted` as their responders drop
     pub(super) async fn close(mut self) {
-        self.context.media.release(&self.scope).await;
         self.pagination_worker.abort();
         let _ = (&mut self.pagination_worker).await;
         self.watcher.abort();
         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;
     }
 
     pub(super) async fn execute(

Modified file: src/engine/watchers.rs+94 additions−44 deletions@@ -1,4 +1,7 @@
-use std::{sync::Arc, time::Duration};
+use std::{
+    sync::Arc,
+    time::{Duration, Instant},
+};
 
 use futures_util::StreamExt;
 use matrix_sdk::{
@@ -20,6 +23,18 @@ use crate::{
 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);
+const SYNC_STABLE: Duration = Duration::from_secs(30);
+
+pub(super) async fn room_subscriptions(
+    sync: Arc<SyncService>,
+    mut rooms: watch::Receiver<Vec<ruma::OwnedRoomId>>,
+) {
+    while rooms.changed().await.is_ok() {
+        let room_ids = rooms.borrow_and_update().clone();
+        let room_ids = room_ids.iter().map(AsRef::as_ref).collect::<Vec<_>>();
+        sync.room_list_service().subscribe_to_rooms(&room_ids).await;
+    }
+}
 
 pub(super) async fn session(
     mut changes: tokio::sync::broadcast::Receiver<SessionChange>,
@@ -57,16 +72,26 @@ pub(super) async fn connection(
     connection: watch::Sender<ConnectionState>,
 ) {
     let mut backoff = SYNC_RESTART_MIN;
+    let mut online_since: Option<Instant> = None;
     while let Some(state) = states.next().await {
+        // only a run that lasted counts as recovered, or a server that answers /versions
+        // but refuses sync would have us bounce between offline and online forever
+        let stable = online_since
+            .take()
+            .map(|since| since.elapsed() >= SYNC_STABLE);
+        if stable == Some(true) {
+            backoff = SYNC_RESTART_MIN;
+        }
         let (state, restart) = match state {
-            SyncState::Idle => (ConnectionState::Idle, false),
+            // only ever us stopping it for a restart, nothing a client needs to hear
+            SyncState::Idle => continue,
             SyncState::Running => {
-                backoff = SYNC_RESTART_MIN;
+                online_since = Some(Instant::now());
                 // 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::Offline => (ConnectionState::Offline, stable == Some(false)),
             SyncState::Terminated => (ConnectionState::Offline, true),
             SyncState::Error(error) => {
                 tracing::warn!(%error, "sync service reported an error");
@@ -80,6 +105,7 @@ pub(super) async fn connection(
             }
         }
         if restart {
+            sync.stop().await;
             tokio::time::sleep(backoff).await;
             backoff = (backoff * 2).min(SYNC_RESTART_MAX);
             sync.start().await;
@@ -132,51 +158,75 @@ pub(super) async fn room_uploads(room: Room, events: EventSender, ready: watch::
     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,
+            Err(RecvError::Closed) => return,
+        };
+        let RoomSendQueueUpdate::MediaUpload {
+            related_to,
+            file,
+            progress,
+            ..
+        } = 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,
+        };
+        if events.send(event).await.is_err() {
+            return;
+        }
+    }
+}
+
+/// failures do reach the client-wide channel, and this one exists before any upload does
+pub(super) async fn upload_failures(
+    client: Client,
+    mut updates: tokio::sync::broadcast::Receiver<matrix_sdk::send_queue::SendQueueUpdate>,
+    events: EventSender,
+) {
+    loop {
+        let update = match updates.recv().await {
+            Ok(update) => update,
             Err(RecvError::Lagged(skipped)) => {
-                tracing::warn!(%room_id, skipped, "upload watcher lagged");
+                tracing::warn!(skipped, "upload failure 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,
+        let 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)
+                })
+            });
+        if !is_attachment {
+            continue;
+        }
+        let event = Event::MediaUploadFailed {
+            room_id: update.room_id,
+            transaction_id,
+            kind: classify_sdk_error(&error),
+            recoverable: is_recoverable,
         };
         if events.send(event).await.is_err() {
             return;

Modified file: src/model.rs+49 additions−5 deletions@@ -322,6 +322,7 @@ pub enum MediaCommand {
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
 pub struct AttachmentUpload {
     pub timeline_id: TimelineId,
     pub filename: String,
@@ -677,7 +678,7 @@ pub enum Event {
     LoggedOut,
     Failed(EngineFailure),
     /// always the last event. keep the saved record unless `LoggedOut` or a
-    /// hard `SessionAccessRejected` came first. commands still in flight get no result
+    /// hard `SessionAccessRejected` came first. commands still in flight answer `Interrupted`
     Stopped,
 }
 
@@ -701,6 +702,8 @@ pub struct RoomSummary {
     pub membership: RoomMembership,
     pub unread_messages: u64,
     pub unread_mentions: u64,
+    /// unread events the push rules said to notify about, which is neither of the above
+    pub unread_notifications: u64,
     pub encrypted: bool,
     pub space: bool,
     pub direct: bool,
@@ -735,12 +738,14 @@ pub struct RoomHero {
     pub avatar: Option<OwnedMxcUri>,
 }
 
-/// enough for a room list preview line. `body` is empty for anything that isn't a plain message
+/// enough for a room list preview line. `kind` and `body` are only set for messages,
+/// and an edit shows its new text
 #[derive(Clone, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[non_exhaustive]
 pub struct LatestEvent {
     pub sender: Option<OwnedUserId>,
+    pub kind: Option<MessageKind>,
     pub body: Option<String>,
     pub timestamp_ms: Option<u64>,
     pub send_state: SendState,
@@ -864,6 +869,8 @@ pub struct RoomPowerLevels {
     pub room_notification: i64,
     pub users: Vec<(OwnedUserId, i64)>,
     pub events: Vec<(String, i64)>,
+    /// in rooms from version 12 the creators outrank everyone and aren't listed in `users`
+    pub creators: Vec<OwnedUserId>,
 }
 
 #[derive(Clone, Debug, Default, Eq, PartialEq)]
@@ -985,6 +992,7 @@ impl RoomMatcher<'_> {
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
 pub enum RoomCategory {
     People,
     Rooms,
@@ -993,10 +1001,13 @@ pub enum RoomCategory {
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
 pub enum MemberFilter {
     Active,
     Joined,
     Invited,
+    Banned,
+    Left,
     All,
 }
 
@@ -1308,7 +1319,7 @@ pub enum TimelineContent {
         display_name: Option<Change<Option<String>>>,
         avatar: Option<Change<Option<OwnedMxcUri>>>,
     },
-    /// `None` means it was cleared
+    /// `None` means it was cleared, or the event was redacted
     RoomName(Option<String>),
     RoomTopic(Option<String>),
     RoomAvatar(Option<OwnedMxcUri>),
@@ -1332,6 +1343,8 @@ pub enum TimelineContent {
 pub struct Message {
     pub kind: MessageKind,
     pub body: String,
+    /// only for `MessageKind::Location`
+    pub geo_uri: Option<String>,
     pub formatted_body: Option<String>,
     pub mentions: Option<MessageMentions>,
     pub edited: bool,
@@ -1414,9 +1427,12 @@ pub struct ThreadSummary {
 #[non_exhaustive]
 pub enum SendState {
     Sent,
-    Queued,
     Sending,
-    Failed { retriable: bool },
+    /// `will_retry` means macltok has it in hand. otherwise it's wedged until you
+    /// `retry_send` or `cancel_send`, so that's when to show a retry button
+    Failed {
+        will_retry: bool,
+    },
 }
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -1463,6 +1479,10 @@ pub struct MediaDescriptor {
     pub size_bytes: Option<u64>,
     pub width: Option<u64>,
     pub height: Option<u64>,
+    pub duration_ms: Option<u64>,
+    pub blurhash: Option<String>,
+    /// the uploader sent a thumbnail, so `MediaSize::Thumbnail` won't need the server to resize
+    pub has_thumbnail: bool,
 }
 
 #[derive(Clone, Debug, Eq, Hash, PartialEq)]
@@ -1480,6 +1500,21 @@ impl MediaId {
     }
 }
 
+/// media ids are opaque but they do have to survive a trip through a browser. a made up one just fails `MediaUnknown`
+impl std::str::FromStr for MediaId {
+    type Err = std::convert::Infallible;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        Ok(Self(value.to_owned()))
+    }
+}
+
+impl From<String> for MediaId {
+    fn from(value: String) -> Self {
+        Self(value)
+    }
+}
+
 impl std::fmt::Display for MediaId {
     fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         self.0.fmt(formatter)
@@ -1488,6 +1523,7 @@ impl std::fmt::Display for MediaId {
 
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
 pub enum MediaSize {
     Full,
     /// a hint. an uploader's own thumbnail wins over the server resizing anything
@@ -1568,7 +1604,10 @@ pub enum RecoveryState {
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
 pub enum VerificationAction {
+    /// take up a request another device sent, which arrives as `VerificationState::Incoming`
+    Accept,
     Confirm,
     Reject,
     Cancel,
@@ -1591,6 +1630,11 @@ pub enum VerificationTarget {
 #[non_exhaustive]
 pub enum VerificationState {
     Waiting,
+    /// someone else asked. `Accept` to go ahead, `Cancel` to turn them down
+    Incoming {
+        user_id: OwnedUserId,
+        device_id: Option<OwnedDeviceId>,
+    },
     Comparison {
         emojis: Vec<(String, String)>,
         decimals: (u16, u16, u16),

Modified file: src/normalise/media.rs+60 additions−78 deletions@@ -76,86 +76,65 @@ pub(super) fn media_identity(item: &SdkTimelineItem, scope: &MediaScope) -> Stri
 }
 
 pub(super) fn media_from_message(identity: &str, message: &MessageType) -> Option<MediaDescriptor> {
-    let (kind, filename, caption, mimetype, size_bytes, width, height) = match message {
-        MessageType::Image(content) => (
-            MediaKind::Image,
-            content.filename().to_owned(),
-            content.caption().map(str::to_owned),
-            content.info.as_ref().and_then(|info| info.mimetype.clone()),
-            content
-                .info
-                .as_ref()
-                .and_then(|info| info.size)
-                .map(u64::from),
-            content
-                .info
-                .as_ref()
-                .and_then(|info| info.width)
-                .map(u64::from),
-            content
-                .info
-                .as_ref()
-                .and_then(|info| info.height)
-                .map(u64::from),
-        ),
-        MessageType::Video(content) => (
-            MediaKind::Video,
-            content.filename().to_owned(),
-            content.caption().map(str::to_owned),
-            content.info.as_ref().and_then(|info| info.mimetype.clone()),
-            content
-                .info
-                .as_ref()
-                .and_then(|info| info.size)
-                .map(u64::from),
-            content
-                .info
-                .as_ref()
-                .and_then(|info| info.width)
-                .map(u64::from),
-            content
-                .info
-                .as_ref()
-                .and_then(|info| info.height)
-                .map(u64::from),
-        ),
-        MessageType::Audio(content) => (
-            MediaKind::Audio,
-            content.filename().to_owned(),
-            content.caption().map(str::to_owned),
-            content.info.as_ref().and_then(|info| info.mimetype.clone()),
-            content
-                .info
-                .as_ref()
-                .and_then(|info| info.size)
-                .map(u64::from),
-            None,
-            None,
-        ),
-        MessageType::File(content) => (
-            MediaKind::File,
-            content.filename().to_owned(),
-            content.caption().map(str::to_owned),
-            content.info.as_ref().and_then(|info| info.mimetype.clone()),
-            content
-                .info
-                .as_ref()
-                .and_then(|info| info.size)
-                .map(u64::from),
-            None,
-            None,
-        ),
-        _ => return None,
-    };
-    Some(MediaDescriptor {
+    let descriptor = |kind, filename: &str, caption: Option<&str>| MediaDescriptor {
         id: media_id(identity, "media"),
         kind,
-        filename,
-        caption,
-        mimetype,
-        size_bytes,
-        width,
-        height,
+        filename: filename.to_owned(),
+        caption: caption.map(str::to_owned),
+        mimetype: None,
+        size_bytes: None,
+        width: None,
+        height: None,
+        duration_ms: None,
+        blurhash: None,
+        has_thumbnail: false,
+    };
+    let millis = |duration: std::time::Duration| u64::try_from(duration.as_millis()).ok();
+    Some(match message {
+        MessageType::Image(content) => {
+            let info = content.info.as_deref();
+            MediaDescriptor {
+                mimetype: info.and_then(|info| info.mimetype.clone()),
+                size_bytes: info.and_then(|info| info.size).map(u64::from),
+                width: info.and_then(|info| info.width).map(u64::from),
+                height: info.and_then(|info| info.height).map(u64::from),
+                blurhash: info.and_then(|info| info.blurhash.clone()),
+                has_thumbnail: info.is_some_and(|info| info.thumbnail_source.is_some()),
+                ..descriptor(MediaKind::Image, content.filename(), content.caption())
+            }
+        }
+        MessageType::Video(content) => {
+            let info = content.info.as_deref();
+            MediaDescriptor {
+                mimetype: info.and_then(|info| info.mimetype.clone()),
+                size_bytes: info.and_then(|info| info.size).map(u64::from),
+                width: info.and_then(|info| info.width).map(u64::from),
+                height: info.and_then(|info| info.height).map(u64::from),
+                duration_ms: info.and_then(|info| info.duration).and_then(millis),
+                blurhash: info.and_then(|info| info.blurhash.clone()),
+                has_thumbnail: info.is_some_and(|info| info.thumbnail_source.is_some()),
+                ..descriptor(MediaKind::Video, content.filename(), content.caption())
+            }
+        }
+        MessageType::Audio(content) => {
+            let info = content.info.as_deref();
+            MediaDescriptor {
+                mimetype: info.and_then(|info| info.mimetype.clone()),
+                size_bytes: info.and_then(|info| info.size).map(u64::from),
+                duration_ms: info.and_then(|info| info.duration).and_then(millis),
+                ..descriptor(MediaKind::Audio, content.filename(), content.caption())
+            }
+        }
+        MessageType::File(content) => {
+            let info = content.info.as_deref();
+            MediaDescriptor {
+                mimetype: info.and_then(|info| info.mimetype.clone()),
+                size_bytes: info.and_then(|info| info.size).map(u64::from),
+                has_thumbnail: info.is_some_and(|info| info.thumbnail_source.is_some()),
+                ..descriptor(MediaKind::File, content.filename(), content.caption())
+            }
+        }
+        _ => return None,
     })
 }
 
@@ -172,6 +151,9 @@ pub(super) fn media_descriptor_for_sticker(
         size_bytes: content.info.size.map(u64::from),
         width: content.info.width.map(u64::from),
         height: content.info.height.map(u64::from),
+        duration_ms: None,
+        blurhash: content.info.blurhash.clone(),
+        has_thumbnail: content.info.thumbnail_source.is_some(),
     }
 }
 

Modified file: src/normalise/mod.rs+1−1

@@ -7,4 +7,4 @@ pub(crate) use media::{MediaRegistration, MediaScope};

77 pub(crate) use member::member_summary;

88 pub(crate) use room::notification_mode;

99 pub(crate) use room::{room_summaries, space_child};

10-pub(crate) use timeline::{normalise_timeline, normalise_timeline_diffs};

10+pub(crate) use timeline::{message_kind, normalise_timeline, normalise_timeline_diffs};

Modified file: src/normalise/room.rs+41 additions−32 deletions@@ -5,14 +5,17 @@ use matrix_sdk::{
     latest_events::LatestEventValue,
     ruma::events::{
         AnyMessageLikeEventContent, AnySyncMessageLikeEvent, AnySyncTimelineEvent, SyncStateEvent,
+        room::message::{Relation, RoomMessageEventContent},
         space::child::SpaceChildEventContent,
     },
 };
 use matrix_sdk_ui::spaces::{SpaceRoom, SpaceService};
 use ruma::OwnedRoomId;
 
+use super::message_kind;
 use crate::{
-    LatestEvent, RoomHero, RoomMembership, RoomNotificationMode, RoomSummary, SendState, SpaceChild,
+    LatestEvent, MessageKind, RoomHero, RoomMembership, RoomNotificationMode, RoomSummary,
+    SendState, SpaceChild,
 };
 
 pub(crate) async fn room_summaries(client: &Client, spaces: &SpaceService) -> Vec<RoomSummary> {
@@ -61,6 +64,7 @@ async fn room_summary(room: &Room, spaces: &SpaceService) -> RoomSummary {
         membership: membership(room.state()),
         unread_messages: room.num_unread_messages(),
         unread_mentions: room.num_unread_mentions(),
+        unread_notifications: room.num_unread_notifications(),
         encrypted: room.encryption_state().is_encrypted(),
         space: room.is_space(),
         direct: room.is_direct().await.unwrap_or(false),
@@ -127,60 +131,65 @@ async fn space_children(room: &Room) -> Vec<OwnedRoomId> {
 }
 
 fn latest_event(room: &Room) -> Option<LatestEvent> {
-    let (sender, body, timestamp, send_state) = match room.latest_event() {
+    let own = || Some(room.own_user_id().to_owned());
+    let local = |local: &matrix_sdk::latest_events::LocalLatestEventValue| {
+        let preview = match local.content.deserialize() {
+            Ok(AnyMessageLikeEventContent::RoomMessage(content)) => preview(&content),
+            _ => (None, None),
+        };
+        (preview, Some(local.timestamp))
+    };
+    let (sender, (kind, body), timestamp, send_state) = match room.latest_event() {
         LatestEventValue::None => return None,
         LatestEventValue::Remote(event) => {
-            let body = match event.raw().deserialize() {
+            let preview = match event.raw().deserialize() {
                 Ok(AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(
                     message,
                 ))) => message
                     .as_original()
-                    .map(|message| message.content.body().to_owned()),
-                _ => None,
+                    .map_or((None, None), |message| preview(&message.content)),
+                _ => (None, None),
             };
-            (event.sender(), body, event.timestamp(), SendState::Sent)
+            (event.sender(), preview, event.timestamp(), SendState::Sent)
         }
         LatestEventValue::RemoteInvite {
             timestamp, inviter, ..
-        } => (inviter, None, Some(timestamp), SendState::Sent),
-        LatestEventValue::LocalIsSending(local) => local_event(room, &local, SendState::Sending),
+        } => (inviter, (None, None), Some(timestamp), SendState::Sent),
+        LatestEventValue::LocalIsSending(value) => {
+            let (preview, timestamp) = local(&value);
+            (own(), preview, timestamp, SendState::Sending)
+        }
         LatestEventValue::LocalHasBeenSent { value, .. } => {
-            local_event(room, &value, SendState::Sent)
+            let (preview, timestamp) = local(&value);
+            (own(), preview, timestamp, SendState::Sent)
         }
-        LatestEventValue::LocalCannotBeSent(local) => {
-            local_event(room, &local, SendState::Failed { retriable: true })
+        // the sdk won't say whether it'll retry, so assume the user has to
+        LatestEventValue::LocalCannotBeSent(value) => {
+            let (preview, timestamp) = local(&value);
+            (
+                own(),
+                preview,
+                timestamp,
+                SendState::Failed { will_retry: false },
+            )
         }
     };
     Some(LatestEvent {
         sender,
+        kind,
         body,
         timestamp_ms: timestamp.map(|timestamp| timestamp.get().into()),
         send_state,
     })
 }
 
-type LatestParts = (
-    Option<ruma::OwnedUserId>,
-    Option<String>,
-    Option<ruma::MilliSecondsSinceUnixEpoch>,
-    SendState,
-);
-
-fn local_event(
-    room: &Room,
-    local: &matrix_sdk::latest_events::LocalLatestEventValue,
-    send_state: SendState,
-) -> LatestParts {
-    let body = match local.content.deserialize() {
-        Ok(AnyMessageLikeEventContent::RoomMessage(content)) => Some(content.body().to_owned()),
-        _ => None,
+/// the latest event of an edited message is the edit, whose own body is the `* fallback`
+fn preview(content: &RoomMessageEventContent) -> (Option<MessageKind>, Option<String>) {
+    let msgtype = match &content.relates_to {
+        Some(Relation::Replacement(replacement)) => &replacement.new_content.msgtype,
+        _ => &content.msgtype,
     };
-    (
-        room.own_user_id().to_owned().into(),
-        body,
-        Some(local.timestamp),
-        send_state,
-    )
+    (Some(message_kind(msgtype)), Some(msgtype.body().to_owned()))
 }
 
 pub(crate) fn space_child(room: SpaceRoom) -> SpaceChild {

Modified file: src/normalise/timeline.rs+19 additions−10 deletions@@ -203,12 +203,12 @@ fn content(content: &TimelineItemContent, media_identity: Option<&str>) -> Timel
             }),
         },
         TimelineItemContent::OtherState(state) => match state.content() {
-            AnyOtherStateEventContentChange::RoomName(change) => {
-                TimelineContent::RoomName(original(change).map(|content| content.name.clone()))
-            }
-            AnyOtherStateEventContentChange::RoomTopic(change) => {
-                TimelineContent::RoomTopic(original(change).map(|content| content.topic.clone()))
-            }
+            AnyOtherStateEventContentChange::RoomName(change) => TimelineContent::RoomName(
+                original(change).and_then(|content| non_empty(&content.name)),
+            ),
+            AnyOtherStateEventContentChange::RoomTopic(change) => TimelineContent::RoomTopic(
+                original(change).and_then(|content| non_empty(&content.topic)),
+            ),
             AnyOtherStateEventContentChange::RoomAvatar(change) => TimelineContent::RoomAvatar(
                 original(change).and_then(|content| content.url.clone()),
             ),
@@ -243,6 +243,11 @@ fn content(content: &TimelineItemContent, media_identity: Option<&str>) -> Timel
     }
 }
 
+/// a cleared name or topic arrives as an empty string
+fn non_empty(value: &str) -> Option<String> {
+    (!value.is_empty()).then(|| value.to_owned())
+}
+
 fn other_state(event_type: String, state_key: &str) -> TimelineContent {
     TimelineContent::OtherState {
         event_type,
@@ -265,6 +270,10 @@ fn msg_like(content: &MsgLikeContent, media_identity: Option<&str>) -> TimelineC
         MsgLikeKind::Message(message) => TimelineContent::Message(Message {
             kind: message_kind(message.msgtype()),
             body: message.body().to_owned(),
+            geo_uri: match message.msgtype() {
+                MessageType::Location(location) => Some(location.geo_uri.clone()),
+                _ => None,
+            },
             formatted_body: formatted_body(message.msgtype()),
             mentions: message.mentions().map(|mentions| MessageMentions {
                 user_ids: mentions.user_ids.iter().cloned().collect(),
@@ -305,7 +314,7 @@ fn msg_like(content: &MsgLikeContent, media_identity: Option<&str>) -> TimelineC
     }
 }
 
-fn message_kind(message: &MessageType) -> MessageKind {
+pub(crate) fn message_kind(message: &MessageType) -> MessageKind {
     match message {
         MessageType::Text(_) => MessageKind::Text,
         MessageType::Emote(_) => MessageKind::Emote,
@@ -379,10 +388,10 @@ fn sender(user_id: &UserId, profile: &TimelineDetails<Profile>) -> Sender {
 fn send_state(state: Option<&EventSendState>) -> SendState {
     match state {
         None | Some(EventSendState::Sent { .. }) => SendState::Sent,
-        Some(EventSendState::NotSentYet { progress: Some(_) }) => SendState::Sending,
-        Some(EventSendState::NotSentYet { progress: None }) => SendState::Queued,
+        Some(EventSendState::NotSentYet { .. }) => SendState::Sending,
+        // recoverable means the sdk parks the queue and we unpark it, nothing for the user to do
         Some(EventSendState::SendingFailed { is_recoverable, .. }) => SendState::Failed {
-            retriable: *is_recoverable,
+            will_retry: *is_recoverable,
         },
     }
 }

Modified file: src/transport.rs+12 additions@@ -719,6 +719,8 @@ impl ClientHandle {
             MemberFilter::Active => RoomMemberships::ACTIVE,
             MemberFilter::Joined => RoomMemberships::JOIN,
             MemberFilter::Invited => RoomMemberships::INVITE,
+            MemberFilter::Banned => RoomMemberships::BAN,
+            MemberFilter::Left => RoomMemberships::LEAVE,
             // An empty mask is the SDK's unfiltered wildcard: it matches every
             // membership, including custom states that `all()` would exclude.
             MemberFilter::All => RoomMemberships::empty(),
@@ -758,6 +760,12 @@ impl ClientHandle {
                 .into_iter()
                 .map(|(event_type, level)| (event_type.to_string(), level.into()))
                 .collect(),
+            creators: levels
+                .rules
+                .privileged_creators
+                .into_iter()
+                .flatten()
+                .collect(),
         })
     }
 
@@ -974,6 +982,10 @@ impl DriverPort {
         self.requests.recv().await
     }
 
+    pub(crate) fn try_recv(&mut self) -> Option<Request> {
+        self.requests.try_recv().ok()
+    }
+
     pub(crate) fn close(&mut self) {
         self.requests.close();
     }

Modified file: src/upload.rs+5 additions−1 deletions@@ -127,7 +127,11 @@ pub(crate) fn prepare_upload(
                 || thumbnail.height == 0
             {
                 return Err(Error::new(
-                    ErrorKind::MediaTooLarge,
+                    if thumbnail.bytes.len() > max_media_bytes {
+                        ErrorKind::MediaTooLarge
+                    } else {
+                        ErrorKind::Invalid
+                    },
                     "thumbnail size or dimensions are invalid",
                 ));
             }

Modified file: tests/media.rs+59 additions@@ -102,6 +102,65 @@ async fn encrypted_media_is_cached_as_plaintext_like_the_sdk_does() {
 }
 
 #[tokio::test]
+async fn a_plain_event_cannot_poison_an_encrypted_files_cache() {
+    use std::io::Read as _;
+
+    let plaintext = b"very secret image".to_vec();
+    let mut cursor = std::io::Cursor::new(plaintext.clone());
+    let mut encryptor = matrix_sdk_crypto::AttachmentEncryptor::new(&mut cursor);
+    let mut ciphertext = Vec::new();
+    encryptor.read_to_end(&mut ciphertext).unwrap();
+    let keys = encryptor.finish();
+    let uri = mxc_uri!("mxc://localhost/secret").to_owned();
+    let file = EncryptedFile::new(uri.clone(), keys.encryption_info, keys.hashes);
+    let mut harness = start(RoomMessageEventContent::new(MessageType::Image(
+        ImageMessageEventContent::plain("bait.png".to_owned(), uri),
+    )))
+    .await;
+    Mock::given(method("GET"))
+        .and(path_regex("/download/localhost/secret$"))
+        .respond_with(ResponseTemplate::new(200).set_body_bytes(ciphertext.clone()))
+        .mount(harness.server.server())
+        .await;
+    assert_eq!(&*fetch(&harness, MediaSize::Full).await, &ciphertext[..]);
+
+    harness
+        .server
+        .sync_room(
+            &harness.client,
+            JoinedRoomBuilder::new(room_id!("!media:localhost")).add_timeline_event(
+                EventFactory::new()
+                    .event(RoomMessageEventContent::new(MessageType::Image(
+                        ImageMessageEventContent::encrypted("secret.png".to_owned(), file),
+                    )))
+                    .sender(&ALICE),
+            ),
+        )
+        .await;
+    let event = support::receive_until(&mut harness.events, |event| {
+        support::event_messages(event).into_iter().any(|message| {
+            support::media_of(message).is_some_and(|media| media.filename == "secret.png")
+        })
+    })
+    .await;
+    let media_id = support::event_messages(&event)
+        .into_iter()
+        .find_map(|message| {
+            support::media_of(message).filter(|media| media.filename == "secret.png")
+        })
+        .map(|media| media.id.clone())
+        .unwrap();
+    let media = harness
+        .handle
+        .media(&media_id, MediaSize::Full)
+        .await
+        .unwrap();
+    assert_eq!(&*media.bytes, &plaintext[..]);
+
+    support::shut_down(&harness.handle, &mut harness.events, harness.running).await;
+}
+
+#[tokio::test]
 async fn simultaneous_fetches_share_one_download() {
     let mut harness = start(plain_image()).await;
     Mock::given(method("GET"))

macltok

Commit 5a52012c9b1d

Commit `5a52012c9b1d`