macltok

public

Star0

Fork 0

Watch0

8 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `fe4133dbcd47`

tarball

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

13 files changed+106-50

engine: say why the engine stopped

Engine::run answered Ok(()) for a shutdown, a logout and a revoked token alike, so a host deciding whether to bin its record and store had to reconstruct the cause from events it might not have seen. A soft logout also needs the device to log back in as, which nothing handed over. Termination names the cause and carries that device id, and Termination::retain turns it into what to keep.

Modified file: README.md+7 additions−5 deletions@@ -38,11 +38,13 @@ separately from its encrypted store. On `SessionTokensChanged`, obtain a new rec
 `ClientHandle::session_record()` and replace the saved record atomically in
 protected storage. Restore with the same store location 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.
+`Engine::run` resolves to a `Termination` saying why it stopped, and
+`Termination::retain` says what to do about it. after `LoggedOut` or
+`AccessRevoked` the record and the store are dead, bin them. `SoftLoggedOut`
+wants the store kept and a fresh login with `PasswordLogin::device_id` set to the
+device it carries, or the e2ee keys go with it. `ShutDown` and `HandlesDropped`
+keep everything. an `Err` is a real failure and kills nothing. 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.
 

Modified file: docs/design-decisions.md+3 additions−1 deletions@@ -41,7 +41,9 @@ 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
+logout and a rejected token both stop the engine. `run` resolves to a
+`Termination` naming which of those it was, or a shutdown, or every handle
+dropping, so the host never works it out from the events. `Shutdown` answers once
 `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

Modified file: examples/headless.rs+7 additions−6 deletions@@ -1,6 +1,6 @@
 use macltok::{
-    EngineConfig, Event, PasswordLogin, RoomMembership, Session, StoreConfig, TimelineContent,
-    TimelineId, TimelineItem,
+    EngineConfig, Event, PasswordLogin, Retain, RoomMembership, Session, StoreConfig,
+    TimelineContent, TimelineId, TimelineItem,
 };
 
 #[tokio::main]
@@ -77,9 +77,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     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,
                     _ => {}
                 }
@@ -91,7 +88,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             }
         }
     }
-    running.await??;
+    match running.await??.retain() {
+        Retain::RecordAndStore => {}
+        Retain::Store => std::fs::remove_file(&record_path)?,
+        Retain::Nothing => std::fs::remove_dir_all(&store_path)?,
+    }
     Ok(())
 }
 

Modified file: src/engine/mod.rs+23 additions−21 deletions@@ -18,7 +18,7 @@ use matrix_sdk::{Client, SendOutsideWasm, SessionChange};
 use matrix_sdk_ui::sync_service::SyncService;
 use ruma::OwnedRoomId;
 use thiserror::Error as ThisError;
-use tokio::sync::{Notify, broadcast, mpsc, watch};
+use tokio::sync::{broadcast, mpsc, watch};
 
 pub(crate) use media::MediaService;
 
@@ -32,7 +32,7 @@ use self::{
 use crate::{
     AccountCommand, ClientHandle, Command, ConnectionState, EncryptionCommand, EngineErrorKind,
     EngineFailure, Error, ErrorKind, Event, EventStream, LifecycleCommand, MediaCommand,
-    NotificationCommand, RoomCommand, TimelineCommand,
+    NotificationCommand, RoomCommand, Termination, TimelineCommand,
     transport::{DriverPort, Request, Responder, transport_pair},
 };
 
@@ -246,7 +246,8 @@ pub struct Engine {
     uploads: HashMap<OwnedRoomId, 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>,
+    stop: mpsc::UnboundedSender<Termination>,
+    stops: mpsc::UnboundedReceiver<Termination>,
 }
 
 impl Engine {
@@ -286,6 +287,7 @@ impl Engine {
         );
         client.send_queue().enable_upload_progress(true);
         let (note_sender, notes) = mpsc::unbounded_channel();
+        let (stop, stops) = 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);
@@ -314,18 +316,19 @@ impl Engine {
             }),
             uploads: HashMap::new(),
             verification_handler: None,
-            stop: Arc::new(Notify::new()),
+            stop,
+            stops,
         };
         Ok((engine, handle, events))
     }
 
     /// 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) -> EngineBoxFuture<'static, Result<(), EngineError>> {
+    pub fn run(self) -> EngineBoxFuture<'static, Result<Termination, EngineError>> {
         Box::pin(self.run_inner())
     }
 
-    async fn run_inner(mut self) -> Result<(), EngineError> {
+    async fn run_inner(mut self) -> Result<Termination, EngineError> {
         let events = self.port.event_sender();
         let user_id = self
             .client
@@ -348,37 +351,32 @@ impl Engine {
         self.resume_upload_forwarders().await;
         self.sync.start().await;
 
-        let mut failure = None;
         let mut shutdown_reply = None;
-        loop {
+        let ending = loop {
             tokio::select! {
                 biased;
-                () = events.closed() => {
-                    failure = Some(EngineError::EventReceiverClosed);
-                    break;
-                }
-                () = self.stop.notified() => break,
+                () = events.closed() => break Err(EngineError::EventReceiverClosed),
+                Some(termination) = self.stops.recv() => break Ok(termination),
                 Some((context, panic)) = self.tasks.join_next(), if !self.tasks.is_empty() => {
                     let Some(message) = panic else { continue };
                     if let TaskContext::Command(label) = context {
                         tracing::error!(label, panic = %message, "command task panicked");
                         continue;
                     }
-                    failure = Some(EngineError::TaskPanicked { task: context.label() });
-                    break;
+                    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 };
+                    let Some(request) = request else { break Ok(Termination::HandlesDropped) };
                     if let Some(reply) = self.dispatch(request).await {
                         shutdown_reply = Some(reply);
-                        break;
+                        break Ok(Termination::ShutDown);
                     }
                 }
             }
-        }
+        };
 
-        if let Some(error) = &failure {
+        if let Err(error) = &ending {
             let _ = events.send(Event::Failed(error.failure())).await;
         }
         self.shutdown().await;
@@ -392,7 +390,7 @@ impl Engine {
                 reply.done(Ok(()));
             }
         }
-        failure.map_or(Ok(()), Err)
+        ending
     }
 
     fn start_tasks(&mut self) {
@@ -422,6 +420,10 @@ impl Engine {
                 Box::pin(watchers::session(
                     queues.session_changes,
                     events(),
+                    client
+                        .device_id()
+                        .expect("authentication was checked during construction")
+                        .to_owned(),
                     self.stop.clone(),
                 )),
             ),
@@ -523,7 +525,7 @@ impl Engine {
                         .await
                         .map_err(|source| Error::from_sdk("logout failed", source))?;
                     let _ = events.send(Event::LoggedOut).await;
-                    stop.notify_one();
+                    let _ = stop.send(Termination::LoggedOut);
                     Ok(())
                 });
             }

Modified file: src/engine/watchers.rs+10 additions−5 deletions@@ -7,12 +7,12 @@ use matrix_sdk::{
     send_queue::{LocalEchoContent, RoomSendQueueUpdate},
 };
 use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};
-use ruma::time::Instant;
-use tokio::sync::{Notify, broadcast::error::RecvError, watch};
+use ruma::{OwnedDeviceId, time::Instant};
+use tokio::sync::{broadcast::error::RecvError, mpsc, watch};
 
 use super::executor::TaskSet;
 use crate::{
-    ConnectionState, Event, encryption::encryption_state, error::classify_sdk_error,
+    ConnectionState, Event, Termination, encryption::encryption_state, error::classify_sdk_error,
     transport::EventSender,
 };
 
@@ -35,7 +35,8 @@ pub(super) async fn room_subscriptions(
 pub(super) async fn session(
     mut changes: tokio::sync::broadcast::Receiver<SessionChange>,
     events: EventSender,
-    stop: Arc<Notify>,
+    device_id: OwnedDeviceId,
+    stop: mpsc::UnboundedSender<Termination>,
 ) {
     loop {
         let event = match changes.recv().await {
@@ -46,7 +47,11 @@ pub(super) async fn session(
                         soft_logout: data.soft_logout,
                     })
                     .await;
-                stop.notify_one();
+                let _ = stop.send(if data.soft_logout {
+                    Termination::SoftLoggedOut { device_id }
+                } else {
+                    Termination::AccessRevoked
+                });
                 return;
             }
             // a missed refresh costs one needless save, a missed rejection comes round again

Modified file: src/lib.rs+3 additions−3 deletions@@ -33,10 +33,10 @@ pub use model::{
     MembershipChange, Message, MessageContent, MessageEditContent, MessageKind, MessageMentions,
     NotificationCommand, OwnDevice, PowerLevelChanges, Profile, PushNotification,
     PushNotificationEvent, PushNotificationRoom, PushNotificationSender, PushNotificationStatus,
-    Reaction, RecoveryState, ReplyDetails, ReplyReference, RichText, RoomCategory, RoomCommand,
-    RoomHero, RoomListFilter, RoomMatcher, RoomMembership, RoomNotificationMode,
+    Reaction, RecoveryState, ReplyDetails, ReplyReference, Retain, RichText, RoomCategory,
+    RoomCommand, RoomHero, RoomListFilter, RoomMatcher, RoomMembership, RoomNotificationMode,
     RoomNotificationSettings, RoomPowerLevels, RoomPreset, RoomSummary, RoomVisibility, SendState,
-    Sender, SpaceChild, ThreadSummary, TimelineCommand, TimelineContent, TimelineDiff,
+    Sender, SpaceChild, Termination, ThreadSummary, TimelineCommand, TimelineContent, TimelineDiff,
     TimelineDiffError, TimelineEvent, TimelineId, TimelineItem, TimelineItemId, TypingState,
     UploadDetails, UploadThumbnail, UtdCause, VerificationAction, VerificationFailure,
     VerificationState, VerificationTarget,

Modified file: src/model.rs+37 additions@@ -682,6 +682,43 @@ pub enum Event {
     Stopped,
 }
 
+/// why `Engine::run` came back, when nothing went wrong
+#[derive(Clone, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+#[non_exhaustive]
+pub enum Termination {
+    ShutDown,
+    /// every `ClientHandle` was dropped, so nobody could have asked for anything else
+    HandlesDropped,
+    LoggedOut,
+    /// the server rejected our token for good
+    AccessRevoked,
+    /// the server wants this device to log in again, and its e2ee keys live in the store
+    SoftLoggedOut {
+        device_id: OwnedDeviceId,
+    },
+}
+
+impl Termination {
+    #[must_use]
+    pub const fn retain(&self) -> Retain {
+        match self {
+            Self::ShutDown | Self::HandlesDropped => Retain::RecordAndStore,
+            Self::SoftLoggedOut { .. } => Retain::Store,
+            Self::LoggedOut | Self::AccessRevoked => Retain::Nothing,
+        }
+    }
+}
+
+/// what's still worth keeping of the saved record and the store
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum Retain {
+    RecordAndStore,
+    Store,
+    Nothing,
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
 #[non_exhaustive]

Modified file: tests/commands.rs+1−1

@@ -130,7 +130,7 @@ async fn start_engine(

130130 MatrixMockServer,

131131 ClientHandle,

132132 EventStream,

133- JoinHandle<Result<(), EngineError>>,

133+ JoinHandle<Result<macltok::Termination, EngineError>>,

134134 ) {

135135 let server = MatrixMockServer::new().await;

136136 server

Modified file: tests/media.rs+1−1

@@ -35,7 +35,7 @@ struct MediaHarness {

3535 client: matrix_sdk::Client,

3636 handle: ClientHandle,

3737 events: EventStream,

38- running: tokio::task::JoinHandle<Result<(), EngineError>>,

38+ running: tokio::task::JoinHandle<Result<macltok::Termination, EngineError>>,

3939 media_id: MediaId,

4040 }

4141

Modified file: tests/send_queue.rs+1−1

@@ -132,7 +132,7 @@ async fn setup() -> (

132132 Client,

133133 ClientHandle,

134134 EventStream,

135- tokio::task::JoinHandle<Result<(), macltok::EngineError>>,

135+ tokio::task::JoinHandle<Result<macltok::Termination, macltok::EngineError>>,

136136 ) {

137137 let server = MatrixMockServer::new().await;

138138 server

Modified file: tests/state.rs+6 additions−2 deletions@@ -2,7 +2,9 @@ mod support;
 
 use std::time::Duration;
 
-use macltok::{ConnectionState, Engine, EngineConfig, EngineError, Event, RoomMembership};
+use macltok::{
+    ConnectionState, Engine, EngineConfig, EngineError, Event, Retain, RoomMembership, Termination,
+};
 use matrix_sdk::{
     ruma::{api::client::sync::sync_events::v5, room_id},
     test_utils::mocks::MatrixMockServer,
@@ -92,7 +94,9 @@ async fn soft_logout_says_so_and_stops_the_engine() {
     })
     .await;
     support::receive_until(&mut events, |event| matches!(event, Event::Stopped)).await;
-    running.await.unwrap().unwrap();
+    let termination = running.await.unwrap().unwrap();
+    assert!(matches!(termination, Termination::SoftLoggedOut { .. }));
+    assert_eq!(termination.retain(), Retain::Store);
 }
 
 #[tokio::test]

Modified file: tests/support/mod.rs+5 additions−2 deletions@@ -24,13 +24,16 @@ pub async fn receive_until(events: &mut EventStream, predicate: impl Fn(&Event) 
 pub async fn shut_down(
     handle: &ClientHandle,
     events: &mut EventStream,
-    running: tokio::task::JoinHandle<Result<(), macltok::EngineError>>,
+    running: tokio::task::JoinHandle<Result<macltok::Termination, macltok::EngineError>>,
 ) {
     let (answer, _) = tokio::join!(handle.shutdown(), async {
         receive_until(events, |event| matches!(event, Event::Stopped)).await
     });
     answer.unwrap();
-    running.await.unwrap().unwrap();
+    assert_eq!(
+        running.await.unwrap().unwrap(),
+        macltok::Termination::ShutDown
+    );
 }
 
 pub fn event_messages(event: &Event) -> Vec<&TimelineEvent> {

Modified file: tests/tier0.rs+2 additions−2 deletions@@ -330,7 +330,7 @@ async fn start_engine() -> (
     MatrixMockServer,
     ClientHandle,
     EventStream,
-    JoinHandle<Result<(), EngineError>>,
+    JoinHandle<Result<macltok::Termination, EngineError>>,
 ) {
     let (server, client) = configured_client().await;
     let (handle, events, running) = run_client(client).await;
@@ -342,7 +342,7 @@ async fn run_client(
 ) -> (
     ClientHandle,
     EventStream,
-    JoinHandle<Result<(), EngineError>>,
+    JoinHandle<Result<macltok::Termination, EngineError>>,
 ) {
     let config = EngineConfig::default().with_initial_history_limit(0);
     let (engine, handle, events) = Engine::from_client(client, config).await.unwrap();

macltok

Commit fe4133dbcd47

Commit `fe4133dbcd47`