macltok

public

Star0

Fork 0

Watch0

2 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `fa9114fcf75f`

tarball

atagen <boss@atagen.co>2026-09-24 13:05 UTC

15 files changed+453-156

engine: fix sync, send queue, media cache and session lifecycle bugs

Modified file: Cargo.toml+1−1

@@ -20,7 +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-tokio = { version = "1.47.1", features = ["rt", "sync"] }

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

2424 tracing = "0.1.41"

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

2626

Modified file: README.md+8 additions−3 deletions@@ -35,9 +35,14 @@ after its fresh `TimelineReset`; the reset precedes the command completion.
 With `sqlite`, persist the initial `Session::record()` separately from its
 encrypted SQLite store. On `SessionTokensChanged`, obtain a new record from
 `ClientHandle::session_record()` and replace the saved record atomically in
-protected storage. Restore with the same store path and passphrase. Delete the
-saved credentials after `LoggedOut` or `SessionAccessRejected`, but keep them
-after `Stopped` so a restart can restore the session. The headless example is
+protected storage. Restore with the same store path and passphrase.
+
+`LoggedOut` and `SessionAccessRejected` both stop the engine. after a logout or a
+hard rejection the record and the store are dead, bin them. a soft logout wants
+the store kept and a fresh login with `PasswordLogin::device_id` set to the old
+device, or the e2ee keys go with it. a bare `Stopped` means keep everything. one
+store directory per login, the sdk won't share a crypto store between devices.
+The headless example is
 login-only; it does not persist sessions across runs.
 
 ## Dependency policy

Modified file: docs/design-decisions.md+4 additions−3 deletions@@ -99,9 +99,10 @@ login, then consume `SessionTokensChanged` and atomically replace that saved
 record with `ClientHandle::session_record()`. A previously captured record may
 hold an exhausted refresh token; failing to persist a rotation must be surfaced
 rather than silently treating the stale record as safe to restore. Restore with
-the same `StoreConfig` path and passphrase. Explicit logout and
-`SessionAccessRejected` invalidate the saved record and store; `Stopped` alone
-does not. Neither cookies nor application-specific secret storage belong in the
+the same `StoreConfig` path and passphrase. logout and a rejected token both
+stop the engine. logout and a hard rejection kill the record and the store, a
+soft logout keeps the store for a same-device login, `Stopped` alone kills
+nothing. Neither cookies nor application-specific secret storage belong in the
 toolkit.
 
 ## Notifications and room data

Modified file: docs/review-2026-09-24.md+14 additions−14 deletions@@ -14,7 +14,7 @@ pass. The sqlite and clippy runs are recorded at the end.
 
 ## C. Correctness
 
-### C1. `Verify(Reject)` never rejects. OPEN
+### C1. `Verify(Reject)` never rejects. done
 
 `VerificationController::reject` (`src/encryption.rs:339-389`) validates the
 SAS state twice and returns `Ok(())` without ever calling `sas.mismatch()`. The
@@ -23,7 +23,7 @@ comparison, and the command reports success. A user who presses "they don't
 match" gets no protection from the one action meant to provide it. No test
 exercises `Reject`.
 
-### C2. Sync dies permanently after `SyncState::Error`. OPEN
+### C2. Sync dies permanently after `SyncState::Error`. done
 
 The connection watcher maps `SyncState::Error` to `ConnectionState::Offline`
 (`src/engine.rs:1338-1341`) and nothing ever calls `SyncService::start` again.
@@ -38,7 +38,7 @@ becomes `ConnectionState::Stopped` while the engine keeps running, so
 Fix direction. Restart with backoff on `Error` and `Terminated`, and report
 `Stopped` only from engine shutdown.
 
-### C3. A room's send queue stays disabled after any send failure. OPEN
+### C3. A room's send queue stays disabled after any send failure. done
 
 The SDK disables a room's queue after every send error, recoverable or not
 (`matrix-sdk send_queue/mod.rs:1011-1019`), and nothing in matrix-sdk-ui turns
@@ -51,7 +51,7 @@ running.
 Fix direction. Re-enable on each transition to `Online`, and keep
 `RetrySend` for unwedging a single item.
 
-### C4. Own encrypted uploads cannot be fetched, and the media cache is split between two formats. OPEN
+### C4. Own encrypted uploads cannot be fetched, and the media cache is split between two formats. done, with a round trip test through the sdk cache
 
 macltok writes ciphertext into the media store under the `Encrypted` source key
 and decrypts on every read (`src/engine.rs:2799-2821`, `2864-2874`). The SDK
@@ -68,7 +68,7 @@ ciphertext as if it were the image. The media tests force plaintext rooms
 Fix direction. Cache plaintext, matching the SDK, and keep the byte bound by
 decrypting in a streaming fashion before the write.
 
-### C5. Media single-flight is poisoned by a leader panic. OPEN
+### C5. Media single-flight is poisoned by a leader panic. folded into the media query rewrite under q1
 
 In `spawn_media_fetch` (`src/engine.rs:1870-1911`) followers push their id and
 return at once, trusting the leader to answer for them. If `fetch_media`
@@ -79,7 +79,7 @@ completes. Largely superseded if media fetch becomes a direct query (see
 M2), but the fix is the same either way. Use a drop guard or a shared future
 so the entry always clears.
 
-### C6. Back pagination can stick at "start reached" forever. OPEN
+### C6. Back pagination can stick at "start reached" forever. done, the sdk already remembers
 
 `reached_start` is a cached `AtomicBool` that is only ever set from a
 pagination result (`src/engine.rs:4060-4071`, `4097`). A limited sync with a
@@ -92,7 +92,7 @@ room scrolls to a stub and stops.
 Fix direction. Drop the flag and follow `Timeline::live_back_pagination_status`
 (`matrix-sdk-ui timeline/pagination.rs:130-138`), which is authoritative.
 
-### C7. Soft logout destroys the store. OPEN
+### C7. Soft logout destroys the store. done, `SessionAccessRejected { soft_logout }` and a same-device login
 
 `SessionChange::UnknownToken` carries `soft_logout`
 (`matrix-sdk client/mod.rs:166`), but the watcher collapses it into a
@@ -102,7 +102,7 @@ store. On a soft logout the correct response is to re-authenticate and keep
 the store, so following the documentation throws away every E2EE key for
 nothing.
 
-### C8. Logout and rejected access leave the engine running. OPEN
+### C8. Logout and rejected access leave the engine running. done
 
 `spawn_logout` emits `LoggedOut` and returns (`src/engine.rs:825-852`). The
 sync service, watchers and room actors keep running against a dead session,
@@ -110,7 +110,7 @@ and every later command fails with an authentication error. Neither logout nor
 a hard `UnknownToken` shuts the engine down. The engine should stop itself,
 emitting `LoggedOut` or the rejection before `Stopped`.
 
-### C9. `MarkRead` can move the read marker backwards and costs two requests. OPEN
+### C9. `MarkRead` can move the read marker backwards and costs two requests. done, open rooms go through the timeline so stale receipts get dropped
 
 `mark_read` (`src/engine.rs:2072-2095`) calls `Room::send_single_receipt`
 twice. That call does no ordering check
@@ -125,14 +125,14 @@ Fix direction. When the room is open, route through the actor and use
 `Timeline::send_multiple_receipts`. Otherwise use a single
 `Room::send_multiple_receipts`.
 
-### C10. `create_direct` always creates a new room. OPEN
+### C10. `create_direct` always creates a new room. done
 
 `Client::create_dm` never reuses an existing DM
 (`matrix-sdk client/mod.rs:1830-1849`). Starting a chat with someone you
 already talk to creates a duplicate every time. `Client::get_dm_room` exists
 for the lookup (`client/mod.rs:1852`).
 
-### C11. Unregistered tasks swallow panics, contradicting the design doc. OPEN
+### C11. Unregistered tasks swallow panics, contradicting the design doc. folded into q1 and the actor restructure
 
 The design doc says a watcher panic is terminal. But the timeline watcher
 (`src/engine.rs:589`), pagination worker (`446`), pagination result task
@@ -143,7 +143,7 @@ account-data writers (`src/engine.rs:1111`, `1141`) and the per-room media
 upload watcher (`1945`) are registered as `TaskContext::Watcher`, so a panic
 in a single write kills the whole engine.
 
-### C12. The room actor blocks on network requests. OPEN
+### C12. The room actor blocks on network requests. done, redact reactions and receipts run beside the actor. replies stay serial so sends keep their order
 
 Each room actor runs its commands serially and awaits each one. Most sends
 only enqueue, but `Timeline::redact` on a remote event is a direct HTTP call
@@ -152,7 +152,7 @@ only enqueue, but `Timeline::redact` on a remote event is a direct HTTP call
 event (`room/reply.rs:108`). A slow homeserver therefore stalls every later
 send, paginate and thread command in that room behind one redaction.
 
-### C13. Media registry evicts media that is still on screen. OPEN
+### C13. Media registry evicts media that is still on screen. done, lru by stamp and released on close
 
 `MediaRegistry::insert_all` (`src/engine.rs:358-370`) evicts in first-insertion
 order and does not refresh position on re-registration. One registry of 2048
@@ -161,7 +161,7 @@ loaded in a room that is still open are the first evicted, and fetching them
 then fails with `MediaUnknown`. Closed timelines never release their entries,
 so they compete with open ones.
 
-### C14. Smaller correctness issues. OPEN
+### C14. Smaller correctness issues. mostly done. `open_room` on an open room waits for q1 snapshots, the lagged watcher stays as is
 
 - `CommandResultError::Failed` renders its `Display` through `Debug`
   (`src/transport.rs:840`), so users see a struct dump.

Modified file: examples/headless.rs+1

@@ -16,6 +16,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {

1616 username,

1717 password,

1818 device_display_name: Some("macltok headless example".into()),

19+ device_id: None,

1920 },

2021 StoreConfig::new(store_path, store_passphrase),

2122 )

Modified file: src/account.rs+5 additions@@ -32,6 +32,8 @@ pub struct PasswordLogin {
     pub username: String,
     pub password: String,
     pub device_display_name: Option<String>,
+    /// set after a soft logout so the store's keys still belong to us
+    pub device_id: Option<OwnedDeviceId>,
 }
 
 pub struct OAuthLogin {
@@ -180,6 +182,9 @@ impl Session {
         if let Some(name) = login.device_display_name.as_deref() {
             request = request.initial_device_display_name(name);
         }
+        if let Some(device_id) = login.device_id.as_deref() {
+            request = request.device_id(device_id.as_str());
+        }
         request
             .send()
             .await

Modified file: src/encryption.rs+3 additions@@ -365,6 +365,9 @@ impl VerificationController {
                 "verification has not reached comparison",
             ));
         }
+        sas.mismatch()
+            .await
+            .map_err(|source| CommandError::from_sdk("rejecting verification failed", source))?;
         // A peer completing the flow concurrently must not be reported as
         // rejected. A remote cancellation racing the send turns the SDK call
         // into a no-op success, so report it instead of a false success.

Modified file: src/engine.rs+310 additions−121 deletions@@ -4,10 +4,8 @@ use std::{
     future::Future,
     io::Read as _,
     panic::AssertUnwindSafe,
-    sync::{
-        Arc,
-        atomic::{AtomicBool, Ordering},
-    },
+    sync::Arc,
+    time::Duration,
 };
 
 use futures_util::{FutureExt, StreamExt};
@@ -154,6 +152,10 @@ type UploadWatchers = HashMap<
         watch::Receiver<Option<Result<(), CommandErrorKind>>>,
     ),
 >;
+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);
+
 type MediaFlights = Arc<Mutex<HashMap<(MediaId, MediaSize), Vec<CommandId>>>>;
 
 pub struct Engine {
@@ -181,6 +183,7 @@ pub struct Engine {
     global_account_gates: HashMap<String, oneshot::Receiver<()>>,
     room_account_gates: HashMap<(OwnedRoomId, String), oneshot::Receiver<()>>,
     media_flights: MediaFlights,
+    stop: Arc<tokio::sync::Notify>,
 }
 
 struct CommandJob<T> {
@@ -340,9 +343,11 @@ enum EngineActivity {
     RoomActor(Option<RoomActorEvent>),
 }
 
+/// least recently seen goes first. stale order slots are skipped by stamp rather than hunted down
 struct MediaRegistry {
-    entries: HashMap<MediaId, MediaRegistration>,
-    order: VecDeque<MediaId>,
+    entries: HashMap<MediaId, (MediaRegistration, u64)>,
+    order: VecDeque<(MediaId, u64)>,
+    next_stamp: u64,
     limit: usize,
 }
 
@@ -351,22 +356,45 @@ impl MediaRegistry {
         Self {
             entries: HashMap::new(),
             order: VecDeque::new(),
+            next_stamp: 0,
             limit,
         }
     }
 
+    fn get(&self, id: &MediaId) -> Option<&MediaRegistration> {
+        self.entries.get(id).map(|(media, _)| media)
+    }
+
     fn insert_all(&mut self, media: Vec<MediaRegistration>) {
         for media in media {
-            if !self.entries.contains_key(&media.id) {
-                self.order.push_back(media.id.clone());
-            }
-            self.entries.insert(media.id.clone(), media);
+            let stamp = self.next_stamp;
+            self.next_stamp += 1;
+            self.order.push_back((media.id.clone(), stamp));
+            self.entries.insert(media.id.clone(), (media, stamp));
         }
         while self.entries.len() > self.limit {
-            if let Some(id) = self.order.pop_front() {
+            let Some((id, stamp)) = self.order.pop_front() else {
+                break;
+            };
+            if self
+                .entries
+                .get(&id)
+                .is_some_and(|(_, current)| *current == stamp)
+            {
                 self.entries.remove(&id);
             }
         }
+        if self.order.len() > self.limit.saturating_mul(4) {
+            let entries = &self.entries;
+            self.order
+                .retain(|(id, stamp)| entries.get(id).is_some_and(|(_, current)| current == stamp));
+        }
+    }
+
+    fn release(&mut self, scope: &MediaScope) {
+        self.entries.retain(|id, _| !scope.owns(id));
+        let entries = &self.entries;
+        self.order.retain(|(id, _)| entries.contains_key(id));
     }
 }
 
@@ -393,6 +421,18 @@ struct RoomActor {
     media: Arc<Mutex<MediaRegistry>>,
     initial_history_limit: u16,
     open_ids: Vec<CommandId>,
+    background: JoinSet<(CommandId, Result<(), CommandError>)>,
+    background_ids: HashMap<TaskId, CommandId>,
+}
+
+type BackgroundResult =
+    Result<(TaskId, (CommandId, Result<(), CommandError>)), tokio::task::JoinError>;
+type BackgroundWork = std::pin::Pin<Box<dyn Future<Output = Result<(), CommandError>> + Send>>;
+
+/// network-bound work runs beside the actor so one slow redaction doesn't queue the room behind it
+enum TimelineOutcome {
+    Done(Result<(), CommandError>),
+    Background(BackgroundWork),
 }
 
 struct TimelineSession {
@@ -408,14 +448,12 @@ struct TimelineSession {
     pagination_worker: JoinHandle<()>,
     pagination_results: Option<mpsc::UnboundedSender<(CommandId, Result<(), CommandError>)>>,
     pagination_result_task: JoinHandle<()>,
-    reached_start: Arc<AtomicBool>,
 }
 
 struct PaginationRequest {
     timeline: Arc<Timeline>,
     events: EventSender,
     timeline_id: TimelineId,
-    reached_start: Arc<AtomicBool>,
     limit: u16,
     command_id: Option<CommandId>,
     results: mpsc::UnboundedSender<(CommandId, Result<(), CommandError>)>,
@@ -491,7 +529,6 @@ impl TimelineSession {
             pagination_ids,
             pagination_results: Some(pagination_results),
             pagination_result_task,
-            reached_start: Arc::new(AtomicBool::new(false)),
         };
         if initial_history_limit > 0 {
             session.paginate(initial_history_limit, None).await;
@@ -522,7 +559,6 @@ impl TimelineSession {
             timeline: self.timeline.clone(),
             events: self.events.clone(),
             timeline_id: self.id.clone(),
-            reached_start: self.reached_start.clone(),
             limit,
             command_id,
             results: self
@@ -550,6 +586,7 @@ impl TimelineSession {
     }
 
     async fn close(mut self, kind: CommandErrorKind, message: &'static str) {
+        self.media.lock().await.release(&self.scope);
         self.pagination_worker.abort();
         let _ = (&mut self.pagination_worker).await;
         drop(self.pagination_results.take());
@@ -678,6 +715,7 @@ impl Engine {
                 global_account_gates: HashMap::new(),
                 room_account_gates: HashMap::new(),
                 media_flights: Arc::new(Mutex::new(HashMap::new())),
+                stop: Arc::new(tokio::sync::Notify::new()),
             },
             handle,
             events,
@@ -718,6 +756,7 @@ impl Engine {
                     terminal_error = Some(EngineError::EventReceiverClosed);
                     break;
                 }
+                () = self.stop.notified() => break,
                 request = self.port.recv() => EngineActivity::Request(request.map(Box::new)),
                 completed = self.tasks.join_next(), if !self.tasks.is_empty() => {
                     EngineActivity::Task(completed)
@@ -825,6 +864,7 @@ impl Engine {
     fn spawn_logout(&mut self, id: CommandId) {
         let events = self.port.event_sender();
         let client = self.client.clone();
+        let stop = self.stop.clone();
         self.tasks.spawn(
             TaskContext::Command {
                 id,
@@ -841,6 +881,7 @@ impl Engine {
                         let _ =
                             emit_command_result(&events, id, CommandKind::Account, Ok(())).await;
                         let _ = events.send(Event::LoggedOut).await;
+                        stop.notify_one();
                     }
                     Err(error) => {
                         let _ = emit_command_result(&events, id, CommandKind::Account, Err(error))
@@ -1306,12 +1347,21 @@ impl Engine {
             .take()
             .expect("session watcher has not started");
         let session_sender = self.port.event_sender();
+        let stop = self.stop.clone();
         self.tasks
             .spawn(TaskContext::Watcher("session change watcher"), async move {
                 loop {
                     let event = match session_changes.recv().await {
                         Ok(SessionChange::TokensRefreshed) => Event::SessionTokensChanged,
-                        Ok(SessionChange::UnknownToken(_)) => Event::SessionAccessRejected,
+                        Ok(SessionChange::UnknownToken(data)) => {
+                            let _ = session_sender
+                                .send(Event::SessionAccessRejected {
+                                    soft_logout: data.soft_logout,
+                                })
+                                .await;
+                            stop.notify_one();
+                            break;
+                        }
                         Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
                             Event::SessionTokensChanged
                         }
@@ -1326,37 +1376,78 @@ impl Engine {
         let mut sync_states = self.sync.state();
         let state_sender = self.port.event_sender();
         let connection_sender = self.port.connection_sender();
+        let sync = self.sync.clone();
+        let client = self.client.clone();
         self.tasks.spawn(
             TaskContext::Watcher("connection state watcher"),
             async move {
+                let mut backoff = SYNC_RESTART_MIN;
                 while let Some(state) = sync_states.next().await {
-                    let state = match state {
-                        SyncState::Idle => ConnectionState::Idle,
-                        SyncState::Running => ConnectionState::Online,
-                        SyncState::Offline => ConnectionState::Offline,
-                        SyncState::Terminated => ConnectionState::Stopped,
+                    let (state, restart) = match state {
+                        SyncState::Idle => (ConnectionState::Idle, false),
+                        SyncState::Running => {
+                            backoff = SYNC_RESTART_MIN;
+                            // 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::Terminated => (ConnectionState::Offline, true),
                         SyncState::Error(error) => {
                             tracing::warn!(error = %error, "sync service reported an error");
-                            ConnectionState::Offline
+                            (ConnectionState::Offline, true)
                         }
                     };
-                    connection_sender.send_replace(state);
-                    if state_sender
-                        .send(Event::ConnectionChanged(state))
-                        .await
-                        .is_err()
-                    {
-                        tracing::debug!(
-                            "connection watcher stopped because the event receiver closed"
-                        );
-                        break;
+                    if *connection_sender.borrow() != state {
+                        connection_sender.send_replace(state);
+                        if state_sender
+                            .send(Event::ConnectionChanged(state))
+                            .await
+                            .is_err()
+                        {
+                            break;
+                        }
+                    }
+                    if restart {
+                        tokio::time::sleep(backoff).await;
+                        backoff = (backoff * 2).min(SYNC_RESTART_MAX);
+                        sync.start().await;
                     }
                 }
-                tracing::debug!("connection watcher stopped because the sync state stream ended");
             },
         );
 
         let client = self.client.clone();
+        let online = self.port.connection_sender().subscribe();
+        self.tasks
+            .spawn(TaskContext::Watcher("send queue watcher"), async move {
+                let mut errors = client.send_queue().subscribe_errors();
+                let mut retries = JoinSet::new();
+                loop {
+                    let error = tokio::select! {
+                        error = errors.recv() => error,
+                        Some(_) = retries.join_next(), if !retries.is_empty() => continue,
+                    };
+                    match error {
+                        Ok(error) if error.is_recoverable => {
+                            let client = client.clone();
+                            let online = online.clone();
+                            retries.spawn(async move {
+                                tokio::time::sleep(SEND_QUEUE_RETRY).await;
+                                if *online.borrow() == ConnectionState::Online
+                                    && let Some(room) = client.get_room(&error.room_id)
+                                {
+                                    room.send_queue().set_enabled(true);
+                                }
+                            });
+                        }
+                        Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
+                        Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+                    }
+                }
+            });
+
+        let client = self.client.clone();
         let encryption_sender = self.port.event_sender();
         self.tasks.spawn(
             TaskContext::Watcher("encryption state watcher"),
@@ -1506,13 +1597,6 @@ impl Engine {
             }
             Command::Room(RoomCommand::Open(room_id)) => self.open_room(id, room_id).await,
             Command::Room(RoomCommand::Close(room_id)) => self.close_room(id, &room_id).await,
-            Command::Room(RoomCommand::MarkRead {
-                timeline_id,
-                event_id,
-            }) => {
-                self.spawn_mark_read(id, timeline_id, event_id);
-                Ok(true)
-            }
             Command::Room(RoomCommand::SetTyping { room_id, typing }) => {
                 self.spawn_typing(id, room_id, typing);
                 Ok(true)
@@ -1646,6 +1730,8 @@ impl Engine {
             media: self.media.clone(),
             initial_history_limit: self.initial_history_limit,
             open_ids,
+            background: JoinSet::new(),
+            background_ids: HashMap::new(),
         };
         let current_user_id = self.client.user_id().map(ToOwned::to_owned);
         self.rooms.insert(
@@ -1718,13 +1804,29 @@ impl Engine {
         id: CommandId,
         command: TimelineCommand,
     ) -> Result<bool, CommandError> {
-        let room_id = command.room_id();
         let sender = self
             .rooms
-            .get(room_id)
+            .get(command.room_id())
             .and_then(RoomActorState::sender)
-            .ok_or_else(|| CommandError::new(CommandErrorKind::RoomNotOpen, "room is not open"))?;
-        send_room_actor_request(sender, RoomActorRequest::Execute { id, command })?;
+            .cloned();
+        match (sender, command) {
+            (Some(sender), command) => {
+                send_room_actor_request(&sender, RoomActorRequest::Execute { id, command })?;
+            }
+            (
+                None,
+                TimelineCommand::MarkRead {
+                    timeline_id,
+                    event_id,
+                },
+            ) => self.spawn_mark_read(id, timeline_id, event_id),
+            (None, _) => {
+                return Err(CommandError::new(
+                    CommandErrorKind::RoomNotOpen,
+                    "room is not open",
+                ));
+            }
+        }
         Ok(true)
     }
 
@@ -2057,7 +2159,10 @@ impl Engine {
         self.verification.lock().await.stop().await;
         self.rooms.clear();
         self.upload_watchers.clear();
-        self.media.lock().await.entries.clear();
+        let mut media = self.media.lock().await;
+        media.entries.clear();
+        media.order.clear();
+        drop(media);
         self.sync.stop().await;
         self.port
             .connection_sender()
@@ -2175,7 +2280,9 @@ async fn retry_send(
     handle.unwedge().await.map_err(|source| {
         CommandError::with_source(CommandErrorKind::Store, "retrying send failed", source)
     })?;
-    client.send_queue().set_enabled(true).await;
+    if let Some(room) = client.get_room(room_id) {
+        room.send_queue().set_enabled(true);
+    }
     Ok(())
 }
 
@@ -2475,11 +2582,14 @@ async fn execute_room_command(
             .await
             .map(|room| Some(room.room_id().to_owned()))
             .map_err(|source| CommandError::from_sdk("creating room failed", source)),
-        RoomCommand::CreateDirect(user_id) => client
-            .create_dm(&user_id)
-            .await
-            .map(|room| Some(room.room_id().to_owned()))
-            .map_err(|source| CommandError::from_sdk("creating direct room failed", source)),
+        RoomCommand::CreateDirect(user_id) => match client.get_dm_room(&user_id) {
+            Some(room) => Ok(Some(room.room_id().to_owned())),
+            None => client
+                .create_dm(&user_id)
+                .await
+                .map(|room| Some(room.room_id().to_owned()))
+                .map_err(|source| CommandError::from_sdk("creating direct room failed", source)),
+        },
         _ => Err(CommandError::new(
             CommandErrorKind::Other,
             "unsupported room command",
@@ -2741,7 +2851,6 @@ async fn fetch_media(
     let media = registry
         .lock()
         .await
-        .entries
         .get(&media_id)
         .cloned()
         .ok_or_else(|| {
@@ -2804,12 +2913,11 @@ async fn fetch_media(
                 "cached media exceeds the configured byte limit",
             ));
         }
-        let bytes = maybe_decrypt_media(cached, &request, max_media_bytes)?;
         events
             .send(Event::MediaReady {
                 media_id,
                 size,
-                bytes: Arc::from(bytes),
+                bytes: Arc::from(cached),
                 mimetype: mimetype.unwrap_or_else(|| "application/octet-stream".into()),
             })
             .await
@@ -2817,8 +2925,9 @@ async fn fetch_media(
         return Ok(());
     }
     let bytes = download_bounded(&client, &request, max_media_bytes).await?;
-    cache_media(&client, &request, &bytes).await;
+    // the sdk caches plaintext under the encrypted key, so we do too or nobody can read anything
     let bytes = maybe_decrypt_media(bytes, &request, max_media_bytes)?;
+    cache_media(&client, &request, &bytes).await;
     events
         .send(Event::MediaReady {
             media_id,
@@ -2927,24 +3036,39 @@ async fn download_bounded(
             "local media is not in the media cache",
         ));
     }
-    let http_request = media_http_request(client, request).await?;
-    let http_client = client.http_client().clone();
-    let request = http_client
-        .execute(request_from_http(http_request)?)
-        .await
-        .map_err(|source| {
-            CommandError::with_source(CommandErrorKind::Network, "media download failed", source)
-        })?;
-    let status = request.status();
-    if status == matrix_sdk::reqwest::StatusCode::UNAUTHORIZED {
+    let mut refreshed = false;
+    let request = loop {
+        let http_request = media_http_request(client, request).await?;
+        let response = client
+            .http_client()
+            .execute(request_from_http(client, http_request)?)
+            .await
+            .map_err(|source| {
+                CommandError::with_source(
+                    CommandErrorKind::Network,
+                    "media download failed",
+                    source,
+                )
+            })?;
+        if response.status() != matrix_sdk::reqwest::StatusCode::UNAUTHORIZED {
+            break response;
+        }
+        if refreshed
+            || client
+                .session_tokens()
+                .is_none_or(|tokens| tokens.refresh_token.is_none())
+        {
+            return Err(CommandError::new(
+                CommandErrorKind::Unauthenticated,
+                "media download was not authorized",
+            ));
+        }
         client.refresh_access_token().await.map_err(|source| {
             CommandError::from_refresh_token("refreshing media access token failed", source)
         })?;
-        return Err(CommandError::new(
-            CommandErrorKind::Unauthenticated,
-            "media download was not authorized",
-        ));
-    }
+        refreshed = true;
+    };
+    let status = request.status();
     if !status.is_success() {
         return Err(CommandError::new(
             CommandErrorKind::MediaUnavailable,
@@ -3175,10 +3299,11 @@ async fn media_http_request(
 }
 
 fn request_from_http(
+    client: &Client,
     request: http::Request<Vec<u8>>,
 ) -> Result<matrix_sdk::reqwest::Request, CommandError> {
     let (parts, _) = request.into_parts();
-    let mut builder = matrix_sdk::reqwest::Client::new().get(parts.uri.to_string());
+    let mut builder = client.http_client().get(parts.uri.to_string());
     for (name, value) in parts.headers.iter() {
         builder = builder.header(name.as_str(), value.as_bytes());
     }
@@ -3188,6 +3313,45 @@ fn request_from_http(
 }
 
 impl RoomActor {
+    async fn finish(
+        &mut self,
+        id: CommandId,
+        outcome: TimelineOutcome,
+    ) -> Result<(), SendEventError> {
+        match outcome {
+            TimelineOutcome::Done(result) => {
+                emit_command_result(&self.events, id, CommandKind::Timeline, result).await
+            }
+            TimelineOutcome::Background(work) => {
+                let task = self.background.spawn(async move { (id, work.await) });
+                self.background_ids.insert(task.id(), id);
+                Ok(())
+            }
+        }
+    }
+
+    async fn reap(&mut self, completed: BackgroundResult) -> Result<(), SendEventError> {
+        let (id, result) = match completed {
+            Ok((task_id, completed)) => {
+                self.background_ids.remove(&task_id);
+                completed
+            }
+            Err(error) => {
+                let Some(id) = self.background_ids.remove(&error.id()) else {
+                    return Ok(());
+                };
+                (
+                    id,
+                    Err(CommandError::new(
+                        CommandErrorKind::Other,
+                        "timeline operation panicked",
+                    )),
+                )
+            }
+        };
+        emit_command_result(&self.events, id, CommandKind::Timeline, result).await
+    }
+
     async fn run(mut self, current_user_id: Option<ruma::OwnedUserId>) {
         let timeline = match self
             .room
@@ -3325,8 +3489,15 @@ impl RoomActor {
         let thread_queue_cap = self.receiver.max_capacity();
         let mut queued_thread_count = 0;
 
+        let mut room_info = self.room.subscribe_info();
         'actor: loop {
             tokio::select! {
+                Some(info) = room_info.next() => {
+                    if matches!(info.state(), matrix_sdk::RoomState::Left | matrix_sdk::RoomState::Banned) {
+                        tracing::debug!(room_id = %self.room_id, "room actor stopped because we left the room");
+                        break 'actor;
+                    }
+                }
                 request = self.receiver.recv() => {
                     let Some(request) = request else {
                         tracing::debug!(room_id = %self.room_id, "room actor stopped because its request channel closed");
@@ -3582,12 +3753,7 @@ impl RoomActor {
                                                 None,
                                             )
                                             .await;
-                                            if emit_command_result(
-                                                &self.events,
-                                                id,
-                                                CommandKind::Timeline,
-                                                result,
-                                            )
+                                            if self.finish(id, result)
                                             .await
                                             .is_err()
                                             {
@@ -3607,12 +3773,7 @@ impl RoomActor {
                                                     thread_root.as_ref(),
                                                 )
                                                 .await;
-                                                if emit_command_result(
-                                                    &self.events,
-                                                    id,
-                                                    CommandKind::Timeline,
-                                                    result,
-                                                )
+                                                if self.finish(id, result)
                                                 .await
                                                 .is_err()
                                                 {
@@ -3744,12 +3905,7 @@ impl RoomActor {
                                                                     thread_root.as_ref(),
                                                                 )
                                                                 .await;
-                                                            if emit_command_result(
-                                                                &self.events,
-                                                                id,
-                                                                CommandKind::Timeline,
-                                                                result,
-                                                            )
+                                                            if self.finish(id, result)
                                                             .await
                                                             .is_err()
                                                             {
@@ -3822,6 +3978,11 @@ impl RoomActor {
                         None => {}
                     }
                 }
+                Some(completed) = self.background.join_next_with_id(), if !self.background.is_empty() => {
+                    if self.reap(completed).await.is_err() {
+                        break 'actor;
+                    }
+                }
                 completed = room_tasks.join_next(), if !room_tasks.is_empty() => {
                     match completed {
                         Some(Err(error)) => {
@@ -3836,6 +3997,9 @@ impl RoomActor {
             }
         }
 
+        while let Some(completed) = self.background.join_next_with_id().await {
+            let _ = self.reap(completed).await;
+        }
         drop(typing_guard);
         room_tasks.abort_all();
         while room_tasks.join_next().await.is_some() {}
@@ -3890,15 +4054,66 @@ async fn execute_timeline_command(
     session: &mut TimelineSession,
     command: TimelineCommand,
     thread_root: Option<&OwnedEventId>,
+) -> TimelineOutcome {
+    let timeline = session.timeline.clone();
+    let work: BackgroundWork = match command {
+        TimelineCommand::Redact {
+            event_id, reason, ..
+        } => Box::pin(async move {
+            timeline
+                .redact(&TimelineEventItemId::EventId(event_id), reason.as_deref())
+                .await
+                .map_err(timeline_error)
+        }),
+        TimelineCommand::ToggleReaction { event_id, key, .. } => Box::pin(async move {
+            timeline
+                .toggle_reaction(&TimelineEventItemId::EventId(event_id), &key)
+                .await
+                .map(|_| ())
+                .map_err(timeline_error)
+        }),
+        TimelineCommand::MarkRead { event_id, .. } => {
+            let thread = thread_root.is_some();
+            Box::pin(async move {
+                let receipt_failed =
+                    |source| CommandError::from_sdk("sending read receipt failed", source);
+                timeline
+                    .send_single_receipt(ReceiptType::Read, event_id.clone())
+                    .await
+                    .map_err(receipt_failed)?;
+                if !thread {
+                    timeline
+                        .send_single_receipt(ReceiptType::FullyRead, event_id)
+                        .await
+                        .map_err(receipt_failed)?;
+                }
+                Ok(())
+            })
+        }
+        command => {
+            return TimelineOutcome::Done(
+                execute_serial_timeline_command(session, command, thread_root).await,
+            );
+        }
+    };
+    TimelineOutcome::Background(work)
+}
+
+async fn execute_serial_timeline_command(
+    session: &mut TimelineSession,
+    command: TimelineCommand,
+    thread_root: Option<&OwnedEventId>,
 ) -> Result<(), CommandError> {
     let timeline = &session.timeline;
     match command {
-        TimelineCommand::OpenThread { .. } | TimelineCommand::CloseThread { .. } => {
-            Err(CommandError::new(
-                CommandErrorKind::Other,
-                "thread lifecycle command was misrouted",
-            ))
-        }
+        TimelineCommand::OpenThread { .. }
+        | TimelineCommand::CloseThread { .. }
+        | TimelineCommand::Redact { .. }
+        | TimelineCommand::ToggleReaction { .. }
+        | TimelineCommand::MarkRead { .. } => Err(CommandError::new(
+            CommandErrorKind::Other,
+            "timeline command was misrouted",
+        )),
         TimelineCommand::Resync { .. } => session.resync().await.map_err(|_| {
             CommandError::new(
                 CommandErrorKind::Timeline,
@@ -4009,17 +4224,6 @@ async fn execute_timeline_command(
                 .await
                 .map_err(timeline_error)
         }
-        TimelineCommand::Redact {
-            event_id, reason, ..
-        } => timeline
-            .redact(&TimelineEventItemId::EventId(event_id), reason.as_deref())
-            .await
-            .map_err(timeline_error),
-        TimelineCommand::ToggleReaction { event_id, key, .. } => timeline
-            .toggle_reaction(&TimelineEventItemId::EventId(event_id), &key)
-            .await
-            .map(|_| ())
-            .map_err(timeline_error),
     }
 }
 
@@ -4051,24 +4255,10 @@ async fn run_pagination(request: PaginationRequest) {
         timeline,
         events,
         timeline_id,
-        reached_start,
         limit,
         command_id,
         results,
     } = request;
-    // One worker per timeline preserves pagination order without accumulating waiting tasks.
-    if reached_start.load(Ordering::Acquire) {
-        let _ = events
-            .send(Event::HistoryChanged {
-                timeline_id,
-                state: HistoryState::StartReached,
-            })
-            .await;
-        if let Some(id) = command_id {
-            let _ = results.send((id, Ok(())));
-        }
-        return;
-    }
     if events
         .send(Event::HistoryChanged {
             timeline_id: timeline_id.clone(),
@@ -4094,7 +4284,6 @@ async fn run_pagination(request: PaginationRequest) {
         .map_err(timeline_error);
     match result {
         Ok(at_start) => {
-            reached_start.store(at_start, Ordering::Release);
             let _ = events
                 .send(Event::HistoryChanged {
                     timeline_id,

Modified file: src/model.rs+19 additions−10 deletions@@ -80,6 +80,7 @@ pub enum RoomCommand {
         level: i64,
     },
     Create(CreateRoomOptions),
+    /// hands back the existing dm if there is one
     CreateDirect(OwnedUserId),
     CancelSend {
         room_id: OwnedRoomId,
@@ -89,10 +90,6 @@ pub enum RoomCommand {
         room_id: OwnedRoomId,
         transaction_id: OwnedTransactionId,
     },
-    MarkRead {
-        timeline_id: TimelineId,
-        event_id: OwnedEventId,
-    },
     SetTyping {
         room_id: OwnedRoomId,
         typing: bool,
@@ -167,6 +164,11 @@ pub enum TimelineCommand {
         event_id: OwnedEventId,
         key: String,
     },
+    /// works on a closed room too, it just can't refuse to move the marker backwards
+    MarkRead {
+        timeline_id: TimelineId,
+        event_id: OwnedEventId,
+    },
 }
 
 impl TimelineCommand {
@@ -179,7 +181,8 @@ impl TimelineCommand {
             | Self::Send { timeline_id, .. }
             | Self::Edit { timeline_id, .. }
             | Self::Redact { timeline_id, .. }
-            | Self::ToggleReaction { timeline_id, .. } => timeline_id.room_id(),
+            | Self::ToggleReaction { timeline_id, .. }
+            | Self::MarkRead { timeline_id, .. } => timeline_id.room_id(),
         }
     }
 
@@ -192,7 +195,8 @@ impl TimelineCommand {
             | Self::Send { timeline_id, .. }
             | Self::Edit { timeline_id, .. }
             | Self::Redact { timeline_id, .. }
-            | Self::ToggleReaction { timeline_id, .. } => Some(timeline_id),
+            | Self::ToggleReaction { timeline_id, .. }
+            | Self::MarkRead { timeline_id, .. } => Some(timeline_id),
         }
     }
 }
@@ -578,8 +582,12 @@ pub enum Event {
     /// Refresh tokens changed. Persist a fresh `ClientHandle::session_record()`
     /// atomically; a record captured before this event may no longer restore.
     SessionTokensChanged,
-    /// Authentication was rejected. Remove the saved session record and store.
-    SessionAccessRejected,
+    /// the server stopped honouring our token and the engine stops. on a soft
+    /// logout keep the store and log in again with the same device, otherwise
+    /// the record and the store are both dead
+    SessionAccessRejected {
+        soft_logout: bool,
+    },
     NotificationSettingsChanged,
     VerificationTargetChanged(VerificationTarget),
     InteractiveAuthRequired {
@@ -637,10 +645,11 @@ pub enum Event {
     },
     /// Terminal failure correlated by `CommandFailure::id`.
     CommandFailed(CommandFailure),
-    /// Explicit logout completed; delete the saved authentication record.
+    /// logout went through and the engine stops. delete the record and the store
     LoggedOut,
     Failed(EngineFailure),
-    /// Engine stopped without logging out; retain saved credentials for restore.
+    /// always the last event. keep the saved record unless `LoggedOut` or a
+    /// hard `SessionAccessRejected` came first. commands still in flight get no result
     Stopped,
 }
 

Modified file: src/normalise/media.rs+6 additions@@ -31,6 +31,12 @@ impl MediaScope {
             nonce: NEXT_MEDIA_SCOPE.fetch_add(1, Ordering::Relaxed),
         }
     }
+
+    pub(crate) fn owns(&self, id: &MediaId) -> bool {
+        id.as_str()
+            .split_once('#')
+            .is_some_and(|(nonce, _)| nonce.parse() == Ok(self.nonce))
+    }
 }
 
 #[derive(Clone)]

Modified file: src/transport.rs+1−1

@@ -837,7 +837,7 @@ pub enum CommandResultError {

837837 Send(#[from] SendCommandError),

838838 #[error("the engine stopped before the operation completed")]

839839 Closed,

840- #[error("{0:?}")]

840+ #[error("{}", .0.message)]

841841 Failed(crate::CommandFailure),

842842 }

843843

Modified file: tests/account.rs+1

@@ -31,6 +31,7 @@ async fn password_session_round_trips_through_an_encrypted_store() {

3131 username: "example".into(),

3232 password: "wordpass".into(),

3333 device_display_name: Some("macltok test".into()),

34+ device_id: None,

3435 },

3536 StoreConfig::new(&store_path, "test store passphrase"),

3637 )

Modified file: tests/media.rs+50 additions−1 deletions@@ -6,11 +6,12 @@ use macltok::{
     TimelineDiff, TimelineItem, UploadDetails, UploadThumbnail,
 };
 use matrix_sdk::{
+    media::{MediaFormat, MediaRequestParameters},
     ruma::{
         api::client::sync::sync_events::v5,
         event_id,
         events::room::{
-            MediaSource,
+            EncryptedFile, MediaSource,
             message::{
                 AudioMessageEventContent, FileInfo, FileMessageEventContent,
                 ImageMessageEventContent, MessageType, RoomMessageEventContent,
@@ -55,6 +56,54 @@ async fn full_size_uses_full_source() {
 }
 
 #[tokio::test]
+async fn encrypted_media_is_cached_as_plaintext_like_the_sdk_does() {
+    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 file = EncryptedFile::new(
+        mxc_uri!("mxc://localhost/secret").to_owned(),
+        keys.encryption_info,
+        keys.hashes,
+    );
+    let source = MediaSource::Encrypted(Box::new(file.clone()));
+    let mut harness = start(RoomMessageEventContent::new(MessageType::Image(
+        ImageMessageEventContent::encrypted("secret.png".to_owned(), file),
+    )))
+    .await;
+    Mock::given(method("GET"))
+        .and(path_regex("/download/localhost/secret$"))
+        .respond_with(ResponseTemplate::new(200).set_body_bytes(ciphertext))
+        .mount(harness.server.server())
+        .await;
+
+    let from_sdk = harness
+        .client
+        .media()
+        .get_media_content(
+            &MediaRequestParameters {
+                source,
+                format: MediaFormat::File,
+            },
+            true,
+        )
+        .await
+        .unwrap();
+    assert_eq!(from_sdk, plaintext);
+    assert_eq!(
+        &*fetch_ready(&mut harness, MediaSize::Full).await,
+        &plaintext[..]
+    );
+    assert_eq!(media_paths(&harness.server).await.len(), 1);
+
+    stop(harness).await;
+}
+
+#[tokio::test]
 async fn simultaneous_fetches_share_one_download() {
     let mut harness = start(plain_image()).await;
     Mock::given(method("GET"))

Modified file: tests/send_queue.rs+6 additions−2 deletions@@ -95,7 +95,11 @@ async fn cancel_queued_attachment_by_pending_send_id() {
 async fn retry_send_wakes_paused_queue() {
     let (server, client, handle, mut events, running) = setup().await;
     let room_id = room_id!("!queue-controls:localhost");
-    client.send_queue().set_enabled(false).await;
+    client
+        .get_room(room_id)
+        .unwrap()
+        .send_queue()
+        .set_enabled(false);
     server
         .mock_room_send()
         .body_matches_partial_json(serde_json::json!({ "body": "retry me" }))
@@ -110,7 +114,7 @@ async fn retry_send_wakes_paused_queue() {
         .await
         .unwrap();
     expect_success(&mut events, id, CommandKind::Room).await;
-    assert!(client.send_queue().is_enabled());
+    assert!(client.get_room(room_id).unwrap().send_queue().is_enabled());
 
     tokio::time::timeout(Duration::from_secs(5), async {
         loop {

Modified file: tests/state.rs+24 additions@@ -83,6 +83,30 @@ async fn connection_watch_tracks_the_engine_lifecycle() {
 }
 
 #[tokio::test]
+async fn soft_logout_says_so_and_stops_the_engine() {
+    let server = MatrixMockServer::new().await;
+    server
+        .mock_sliding_sync()
+        .error_unknown_token(true)
+        .mount()
+        .await;
+    let client = server.client_builder().build().await;
+    let (engine, _handle, mut events) = Engine::from_client(
+        client,
+        EngineConfig::default().with_initial_history_limit(0),
+    )
+    .await
+    .unwrap();
+    let running = tokio::spawn(engine.run());
+    receive_until(&mut events, |event| {
+        matches!(event, Event::SessionAccessRejected { soft_logout: true })
+    })
+    .await;
+    receive_until(&mut events, |event| matches!(event, Event::Stopped)).await;
+    running.await.unwrap().unwrap();
+}
+
+#[tokio::test]
 async fn closing_the_event_stream_stops_an_idle_engine() {
     let server = MatrixMockServer::new().await;
     server

macltok

Commit fa9114fcf75f

Commit `fa9114fcf75f`