macltok

public

Star0

Fork 0

Watch0

19 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `673563a76e43`

tarball

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

2 files changed+75-20

engine: let a logout finish and say so

A logout revokes its own token, so the sync it races can hear the rejection first and end the engine as AccessRevoked with the logout aborted halfway. A rejection that turns up while a logout is in flight is now held for that logout, and only speaks if the logout fails. Dropping every handle waits for a logout in flight rather than aborting it, and the stop channel is read before the closed event stream so a host that drops the stream straight after LoggedOut still learns how it ended.

Modified file: src/engine/mod.rs+46 additions−11 deletions@@ -16,7 +16,7 @@ use std::{
 use futures_util::FutureExt;
 use matrix_sdk::{Client, SendOutsideWasm, SessionChange};
 use matrix_sdk_ui::sync_service::SyncService;
-use ruma::OwnedRoomId;
+use ruma::{OwnedDeviceId, OwnedRoomId};
 use thiserror::Error as ThisError;
 use tokio::sync::{broadcast, mpsc, watch};
 
@@ -214,6 +214,13 @@ enum Deferred {
     Close(Responder),
 }
 
+/// logouts in flight, and a token rejection held back because one of them caused it
+#[derive(Default)]
+pub(super) struct Logouts {
+    pub(super) in_flight: usize,
+    pub(super) rejected: Option<Option<OwnedDeviceId>>,
+}
+
 struct Queues {
     mutations: mpsc::Receiver<MutationJob>,
     notifications: mpsc::Receiver<Job<NotificationCommand>>,
@@ -250,6 +257,7 @@ pub struct Engine {
     verification_handler: Option<matrix_sdk::event_handler::EventHandlerHandle>,
     stop: mpsc::UnboundedSender<Termination>,
     stops: mpsc::UnboundedReceiver<Termination>,
+    logouts: Arc<watch::Sender<Logouts>>,
 }
 
 impl Engine {
@@ -323,6 +331,7 @@ impl Engine {
             verification_handler: None,
             stop,
             stops,
+            logouts: Arc::new(watch::Sender::new(Logouts::default())),
         };
         Ok((engine, handle, events))
     }
@@ -357,11 +366,16 @@ impl Engine {
         self.sync.start().await;
 
         let mut shutdown_reply = None;
+        let mut handles_dropped = false;
+        let mut logouts = self.logouts.subscribe();
         let ending = loop {
             tokio::select! {
                 biased;
-                () = events.closed() => break Err(EngineError::EventReceiverClosed),
                 Some(termination) = self.stops.recv() => break Ok(termination),
+                () = events.closed() => break Err(EngineError::EventReceiverClosed),
+                () = settled(&mut logouts), if handles_dropped => {
+                    break Ok(Termination::HandlesDropped);
+                }
                 Some((context, panic)) = self.tasks.join_next(), if !self.tasks.is_empty() => {
                     let Some(message) = panic else { continue };
                     if let TaskContext::Command(label) = context {
@@ -371,8 +385,12 @@ impl Engine {
                     break Err(EngineError::TaskPanicked { task: context.label() });
                 }
                 Some(note) = self.notes.recv() => self.actor_note(note).await,
-                request = self.port.recv() => {
-                    let Some(request) = request else { break Ok(Termination::HandlesDropped) };
+                request = self.port.recv(), if !handles_dropped => {
+                    // nobody's left to ask, but a logout they asked for still deserves to land
+                    let Some(request) = request else {
+                        handles_dropped = true;
+                        continue;
+                    };
                     if let Some(reply) = self.dispatch(request).await {
                         shutdown_reply = Some(reply);
                         break Ok(Termination::ShutDown);
@@ -431,6 +449,7 @@ impl Engine {
                         .expect("authentication was checked during construction")
                         .to_owned(),
                     self.stop.clone(),
+                    self.logouts.clone(),
                 )),
             ),
             (
@@ -525,14 +544,26 @@ impl Engine {
             Command::Account(AccountCommand::Logout) => {
                 let events = self.port.event_sender();
                 let stop = self.stop.clone();
+                let logouts = self.logouts.clone();
+                logouts.send_modify(|logouts| logouts.in_flight += 1);
                 self.spawn_command("logout", reply, async move {
-                    client
-                        .logout()
-                        .await
-                        .map_err(|source| Error::from_sdk("logout failed", source))?;
-                    let _ = events.send(Event::LoggedOut).await;
-                    let _ = stop.send(Termination::LoggedOut);
-                    Ok(())
+                    let result = client.logout().await;
+                    if result.is_ok() {
+                        let _ = events.send(Event::LoggedOut).await;
+                        let _ = stop.send(Termination::LoggedOut);
+                    }
+                    let mut rejected = None;
+                    logouts.send_if_modified(|logouts| {
+                        if result.is_err() && logouts.in_flight == 1 {
+                            rejected = logouts.rejected.take();
+                        }
+                        false
+                    });
+                    if let Some(soft_logout) = rejected {
+                        watchers::reject_session(&events, &stop, soft_logout).await;
+                    }
+                    logouts.send_modify(|logouts| logouts.in_flight -= 1);
+                    result.map_err(|source| Error::from_sdk("logout failed", source))
                 });
             }
             Command::Account(AccountCommand::SetAccountData {
@@ -883,6 +914,10 @@ fn defer(pending: &mut VecDeque<Deferred>, deferred: Deferred, capacity: usize) 
     )));
 }
 
+async fn settled(logouts: &mut watch::Receiver<Logouts>) {
+    let _ = logouts.wait_for(|logouts| logouts.in_flight == 0).await;
+}
+
 fn enqueue<T>(sender: &mpsc::Sender<Job<T>>, job: Job<T>) {
     if let Err(error) = sender.try_send(job) {
         let (job, error) = match error {

Modified file: src/engine/watchers.rs+29 additions−9 deletions@@ -10,7 +10,7 @@ use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};
 use ruma::{OwnedDeviceId, OwnedRoomId, OwnedTransactionId, time::Instant};
 use tokio::sync::{broadcast::error::RecvError, mpsc, watch};
 
-use super::executor::TaskSet;
+use super::{Logouts, executor::TaskSet};
 use crate::{
     ConnectionState, Event, Termination, UploadOutcome, encryption::encryption_state,
     error::classify_sdk_error, transport::EventSender,
@@ -37,21 +37,25 @@ pub(super) async fn session(
     events: EventSender,
     device_id: OwnedDeviceId,
     stop: mpsc::UnboundedSender<Termination>,
+    logouts: Arc<watch::Sender<Logouts>>,
 ) {
     loop {
         let event = match changes.recv().await {
             Ok(SessionChange::TokensRefreshed) => Event::SessionTokensChanged,
             Ok(SessionChange::UnknownToken(data)) => {
                 let soft_logout = data.soft_logout.then_some(device_id);
-                let _ = events
-                    .send(Event::SessionAccessRejected {
-                        soft_logout: soft_logout.clone(),
-                    })
-                    .await;
-                let _ = stop.send(match soft_logout {
-                    Some(device_id) => Termination::SoftLoggedOut { device_id },
-                    None => Termination::AccessRevoked,
+                let mut held = false;
+                // our own logout revokes the token, so a rejection meanwhile is its doing
+                logouts.send_if_modified(|logouts| {
+                    held = logouts.in_flight > 0;
+                    if held {
+                        logouts.rejected = Some(soft_logout.clone());
+                    }
+                    held
                 });
+                if !held {
+                    reject_session(&events, &stop, soft_logout).await;
+                }
                 return;
             }
             // a missed refresh costs one needless save, a missed rejection comes round again
@@ -64,6 +68,22 @@ pub(super) async fn session(
     }
 }
 
+pub(super) async fn reject_session(
+    events: &EventSender,
+    stop: &mpsc::UnboundedSender<Termination>,
+    soft_logout: Option<OwnedDeviceId>,
+) {
+    let _ = events
+        .send(Event::SessionAccessRejected {
+            soft_logout: soft_logout.clone(),
+        })
+        .await;
+    let _ = stop.send(match soft_logout {
+        Some(device_id) => Termination::SoftLoggedOut { device_id },
+        None => Termination::AccessRevoked,
+    });
+}
+
 /// `states` must be taken before sync starts, a late subscriber never hears the transition it missed
 pub(super) async fn connection(
     mut states: impl futures_util::Stream<Item = SyncState> + Unpin,

macltok

Commit 673563a76e43

Commit `673563a76e43`