macltok

public

Star0

Fork 0

Watch0

20 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `f12377bce9b9`

tarball

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

11 files changed+450-112

engine: keep the session record saved as tokens change

A refresh token works once, so a host that misses one save after SessionTokensChanged has lost the login, and every host had to get that right on a bounded stream it may be slow to drain. The Session now carries its SessionStore into the engine, which saves each rotation off the event stream before announcing it. The session changes are subscribed when the client is built, because the sdk's initialisation task can refresh during restore before any engine exists. Natively the sdk's save callback writes the record inside the refresh itself through SessionStore::save_blocking, which closes the crash window before the new token is used.

Modified file: README.md+9 additions−3 deletions@@ -44,9 +44,15 @@ lands. stop applying, call `resync`, and pick up from the next `TimelineReset`
 with `sqlite` or `indexeddb`, a `SessionStore` holds one login's record and its
 store. `EncryptedSessionStore` seals the record with the store's passphrase, in a
 file you pick natively or an IndexedDB database beside the store in a browser.
-logging in saves the first record and `Session::restore` loads it. on every
-`SessionTokensChanged`, `save` a fresh `ClientHandle::session_record()` before
-the next refresh, refresh tokens only work once
+logging in saves the first record and `Session::restore` loads it. from then on
+the engine the `Session` builds keeps it current. every token rotation is saved
+before `SessionTokensChanged` goes out, and a save that fails says so with
+`SessionSaveFailed`. natively `EncryptedSessionStore` writes from inside the
+sdk's refresh, before the new refresh token is ever used, and a store of your
+own gets the same by implementing `SessionStore::save_blocking`. in a browser
+the save can only follow the refresh, so a tab killed in between wakes with a
+spent token. `Session::without_persistence` hands the job back, and then it's
+yours to save `ClientHandle::session_record()` on every `SessionTokensChanged`
 
 `Engine::run` resolves to a `Termination` saying why it stopped, and
 `Termination::retain` says what to do about it. after `LoggedOut` or

Modified file: docs/design-decisions.md+11 additions−2 deletions@@ -168,8 +168,17 @@ with an encrypted store, refresh tokens and threads. the store is sqlite
 natively and indexeddb in a browser. secret-bearing inputs don't derive `Clone`
 
 a `SessionStore` owns one login's record and its store. login saves the first
-record, `restore` loads it, and the caller saves `ClientHandle::session_record()`
-on every `SessionTokensChanged`. `EncryptedSessionStore` seals the record with
+record, `restore` loads it, and the engine a `Session` builds saves every token
+rotation after that, since a refresh token works once and a host that missed one
+save has lost the login. the session changes are subscribed the moment the
+client exists, because the sdk can refresh during restore before any engine
+does. a save runs off the event stream, so a full stream never holds it back,
+and `SessionTokensChanged` only goes out once it has landed, or
+`SessionSaveFailed` if it didn't. natively the sdk's save callback runs inside
+the refresh, before the new token is used, and `save_blocking` writes there and
+then. in a browser there's only the async save, so a tab that dies between a
+refresh and its save wakes up holding a spent token. `without_persistence`
+hands the job back to the host. `EncryptedSessionStore` seals the record with
 the sdk's `StoreCipher` under the store's passphrase and replaces it by rename,
 so a crash mid save leaves the old one. it keeps the record out of the sdk's
 state store because sqlite writes custom values there in the clear, and a

Modified file: examples/headless.rs+11 additions−9 deletions@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use macltok::{
     EncryptedSessionStore, EngineConfig, Event, PasswordLogin, Restore, Retain, RoomMembership,
     Session, SessionStore, StoreConfig, TimelineContent, TimelineDiff, TimelineId, TimelineItem,
@@ -8,15 +10,18 @@ use macltok::{
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let store_path = std::path::PathBuf::from(required("MATRIX_STORE_PATH")?);
     let passphrase = required("MATRIX_STORE_PASSPHRASE")?;
-    let sessions = EncryptedSessionStore::file(
+    let sessions = Arc::new(EncryptedSessionStore::file(
         store_path.with_extension("session"),
         StoreConfig::sqlite(&store_path, &passphrase),
-    );
-    let session = match Session::restore(&sessions).await? {
+    ));
+    let session = match Session::restore(sessions.clone()).await? {
         Some(Restore::Resumed(session)) => session,
         Some(Restore::SoftLoggedOut(soft)) => {
-            Session::login_password(soft.password_login(required("MATRIX_PASSWORD")?), &sessions)
-                .await?
+            Session::login_password(
+                soft.password_login(required("MATRIX_PASSWORD")?),
+                sessions.clone(),
+            )
+            .await?
         }
         None => {
             Session::login_password(
@@ -26,7 +31,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     required("MATRIX_PASSWORD")?,
                 )
                 .with_device_display_name("macltok headless example"),
-                &sessions,
+                sessions.clone(),
             )
             .await?
         }
@@ -78,9 +83,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                         }
                         print_messages(&items);
                     }
-                    Event::SessionTokensChanged => {
-                        sessions.save(&handle.session_record()?).await?;
-                    }
                     Event::Stopped => break,
                     _ => {}
                 }

Modified file: src/account.rs+122 additions−15 deletions@@ -1,9 +1,12 @@
 #[cfg(feature = "sqlite")]
 use std::path::PathBuf;
+#[cfg(not(target_family = "wasm"))]
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::{collections::HashMap, sync::Arc};
 
 use matrix_sdk::{
-    Client, ClientBuilder, SendOutsideWasm, SessionMeta, SyncOutsideWasm, ThreadingSupport,
+    Client, ClientBuilder, SendOutsideWasm, SessionChange, SessionMeta, SyncOutsideWasm,
+    ThreadingSupport,
     authentication::{
         AuthSession,
         matrix::MatrixSession,
@@ -18,14 +21,27 @@ use matrix_sdk::{
 };
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
+use tokio::sync::{broadcast, mpsc};
 use url::Url;
 
 use crate::{
-    ClientHandle, Engine, EngineConfig, EngineError, EventStream, SessionStore, error::SourceError,
+    ClientHandle, Engine, EngineConfig, EngineError, ErrorKind, EventStream, SessionStore,
+    engine::Persistence, error::SourceError,
 };
 
+/// a logged in client and the `SessionStore` its record lives in. the engine it builds
+/// saves every token rotation there, unless told `without_persistence`
 pub struct Session {
     client: Client,
+    tracking: Tracking,
+}
+
+/// taken as soon as the client exists, since the sdk can refresh before there's an engine
+struct Tracking {
+    session_changes: broadcast::Receiver<SessionChange>,
+    persistence: Option<Persistence>,
+    #[cfg(not(target_family = "wasm"))]
+    inline: Arc<AtomicBool>,
 }
 
 #[non_exhaustive]
@@ -127,6 +143,7 @@ pub enum OAuthApplicationType {
 pub struct PendingOAuthLogin {
     client: Client,
     sessions: Arc<dyn SessionStore>,
+    tracking: Option<Tracking>,
     authorization_url: String,
     state: Option<CsrfToken>,
 }
@@ -164,7 +181,11 @@ impl PendingOAuthLogin {
             oauth.abort_login(&state).await;
             return Err(account_error("OAuth login", error));
         }
-        Session::saved(self.client.clone(), &*self.sessions).await
+        let tracking = self
+            .tracking
+            .take()
+            .ok_or(AccountError::OAuthCallbackAlreadySubmitted)?;
+        Session::saved(self.client.clone(), &*self.sessions, tracking).await
     }
 
     pub async fn abort(mut self) {
@@ -239,8 +260,8 @@ impl StoreConfig {
     }
 }
 
-/// Serializable authentication secrets. A `SessionStore` keeps it, and it must
-/// be replaced when `Event::SessionTokensChanged` arrives.
+/// Serializable authentication secrets. A `SessionStore` keeps it, and the engine
+/// replaces it there on every token rotation.
 #[derive(Clone, Deserialize, Serialize)]
 pub struct SessionRecord {
     homeserver: String,
@@ -334,13 +355,14 @@ pub enum AccountError {
 impl Session {
     pub async fn login_password(
         login: PasswordLogin,
-        sessions: &dyn SessionStore,
+        sessions: Arc<dyn SessionStore>,
     ) -> Result<Self, AccountError> {
         let client = matrix_client_builder(sessions.store())
             .server_name_or_homeserver_url(login.homeserver.trim())
             .build()
             .await
             .map_err(|error| account_error("building the Matrix client", error))?;
+        let tracking = Tracking::start(&client, sessions.clone())?;
         let mut request = client
             .matrix_auth()
             .login_username(login.username.trim(), &login.password)
@@ -355,7 +377,7 @@ impl Session {
             .send()
             .await
             .map_err(|error| account_error("password login", error))?;
-        Self::saved(client, sessions).await
+        Self::saved(client, &*sessions, tracking).await
     }
 
     pub async fn begin_oauth(
@@ -371,6 +393,7 @@ impl Session {
             .build()
             .await
             .map_err(|error| account_error("building the Matrix client", error))?;
+        let tracking = Tracking::start(&client, sessions.clone())?;
         let application_type = match login.application_type {
             OAuthApplicationType::Native => ApplicationType::Native,
             OAuthApplicationType::Web => ApplicationType::Web,
@@ -401,14 +424,15 @@ impl Session {
         Ok(PendingOAuthLogin {
             client,
             sessions,
+            tracking: Some(tracking),
             authorization_url: authorization.url.to_string(),
             state: Some(authorization.state),
         })
     }
 
     /// Restore the record `sessions` holds, or `None` when it holds none.
-    /// The caller must also handle refresh, rejection, and explicit logout.
-    pub async fn restore(sessions: &dyn SessionStore) -> Result<Option<Restore>, AccountError> {
+    /// The caller must also handle rejection and explicit logout.
+    pub async fn restore(sessions: Arc<dyn SessionStore>) -> Result<Option<Restore>, AccountError> {
         let Some(record) = sessions.load().await? else {
             return Ok(None);
         };
@@ -431,28 +455,111 @@ impl Session {
             .build()
             .await
             .map_err(|error| account_error("reopening the Matrix store", error))?;
+        let tracking = Tracking::start(&client, sessions)?;
         client
             .restore_session(authentication)
             .await
             .map_err(|error| account_error("restoring the authenticated session", error))?;
-        Ok(Some(Restore::Resumed(Self { client })))
+        Ok(Some(Restore::Resumed(Self { client, tracking })))
+    }
+
+    /// leave the record to the host, which then saves `ClientHandle::session_record()`
+    /// on every `SessionTokensChanged`
+    #[must_use]
+    pub fn without_persistence(mut self) -> Self {
+        self.tracking.persistence = None;
+        #[cfg(not(target_family = "wasm"))]
+        self.tracking.inline.store(false, Ordering::Release);
+        self
     }
 
     pub async fn build(
         self,
         config: EngineConfig,
     ) -> Result<(Engine, ClientHandle, EventStream), AccountError> {
-        Engine::from_client(self.client, config)
-            .await
-            .map_err(AccountError::from)
+        Engine::with_persistence(
+            self.client,
+            config,
+            self.tracking.session_changes,
+            self.tracking.persistence,
+        )
+        .await
+        .map_err(AccountError::from)
     }
 
-    async fn saved(client: Client, sessions: &dyn SessionStore) -> Result<Self, AccountError> {
+    async fn saved(
+        client: Client,
+        sessions: &dyn SessionStore,
+        tracking: Tracking,
+    ) -> Result<Self, AccountError> {
         sessions.save(&session_record(&client)?).await?;
-        Ok(Self { client })
+        Ok(Self { client, tracking })
+    }
+}
+
+impl Tracking {
+    fn start(client: &Client, sessions: Arc<dyn SessionStore>) -> Result<Self, AccountError> {
+        let session_changes = client.subscribe_to_session_changes();
+        #[cfg_attr(target_family = "wasm", allow(unused_variables))]
+        let (results, inline_results) = mpsc::unbounded_channel();
+        #[cfg(not(target_family = "wasm"))]
+        let inline = save_inline(client, sessions.clone(), results)?;
+        Ok(Self {
+            session_changes,
+            persistence: Some(Persistence {
+                save: Box::new(move |client| {
+                    let sessions = sessions.clone();
+                    Box::pin(async move {
+                        let record = session_record(&client).map_err(save_failed)?;
+                        sessions.save(&record).await.map_err(save_failed)
+                    })
+                }),
+                inline: inline_results,
+            }),
+            #[cfg(not(target_family = "wasm"))]
+            inline,
+        })
     }
 }
 
+/// the sdk calls this inside its refresh, after it has the new tokens and before it uses
+/// them, so a store that writes there and then can't lose a refresh token to a crash
+#[cfg(not(target_family = "wasm"))]
+fn save_inline(
+    client: &Client,
+    sessions: Arc<dyn SessionStore>,
+    results: mpsc::UnboundedSender<Result<(), crate::Error>>,
+) -> Result<Arc<AtomicBool>, AccountError> {
+    let enabled = Arc::new(AtomicBool::new(true));
+    let saving = enabled.clone();
+    client
+        .set_session_callbacks(
+            // only asked for with the cross process refresh lock, which we never enable
+            Box::new(|client| {
+                client
+                    .session_tokens()
+                    .ok_or_else(|| AccountError::MissingSession.into())
+            }),
+            Box::new(move |client| {
+                if !saving.load(Ordering::Acquire) {
+                    return Ok(());
+                }
+                let record = session_record(&client)?;
+                if let Some(result) = sessions.save_blocking(&record) {
+                    let _ = results.send(result.map_err(save_failed));
+                }
+                Ok(())
+            }),
+        )
+        .map_err(|error| account_error("registering the session callbacks", error))?;
+    Ok(enabled)
+}
+
+fn save_failed(error: AccountError) -> crate::Error {
+    let message = format!("saving the session record failed, {error}");
+    crate::Error::with_source(ErrorKind::Store, message, error)
+}
+
 pub(crate) fn session_record(client: &Client) -> Result<SessionRecord, AccountError> {
     let authentication = match client.session().ok_or(AccountError::MissingSession)? {
         AuthSession::Matrix(session) => AuthenticationRecord::Matrix(session),

Modified file: src/engine/mod.rs+31 additions−6 deletions@@ -3,6 +3,7 @@ pub(crate) mod executor;
 mod media;
 mod mutations;
 mod room;
+mod session;
 mod timeline;
 mod watchers;
 
@@ -21,6 +22,7 @@ use thiserror::Error as ThisError;
 use tokio::sync::{broadcast, mpsc, watch};
 
 pub(crate) use media::MediaService;
+pub(crate) use session::Persistence;
 
 use self::{
     commands::Job,
@@ -226,6 +228,7 @@ struct Queues {
     notifications: mpsc::Receiver<Job<NotificationCommand>>,
     encryption: mpsc::Receiver<Job<EncryptionCommand>>,
     session_changes: broadcast::Receiver<SessionChange>,
+    persistence: Option<Persistence>,
     upload_reports: mpsc::Receiver<watchers::RoomUpload>,
 }
 
@@ -266,6 +269,18 @@ impl Engine {
         client: Client,
         config: EngineConfig,
     ) -> Result<(Self, ClientHandle, EventStream), EngineError> {
+        let session_changes = client.subscribe_to_session_changes();
+        Self::with_persistence(client, config, session_changes, None).await
+    }
+
+    /// `session_changes` is taken when the client is built, since the sdk can refresh
+    /// before an engine exists
+    pub(crate) async fn with_persistence(
+        client: Client,
+        config: EngineConfig,
+        session_changes: broadcast::Receiver<SessionChange>,
+        persistence: Option<Persistence>,
+    ) -> Result<(Self, ClientHandle, EventStream), EngineError> {
         if client.user_id().is_none() {
             return Err(EngineError::Unauthenticated);
         }
@@ -275,7 +290,6 @@ impl Engine {
         {
             return Err(EngineError::ZeroCapacity);
         }
-        let session_changes = client.subscribe_to_session_changes();
         let sync = Arc::new(
             SyncService::builder(client.clone())
                 .with_offline_mode()
@@ -324,6 +338,7 @@ impl Engine {
                 notifications: notification_queue,
                 encryption: encryption_queue,
                 session_changes,
+                persistence,
                 upload_reports: upload_report_queue,
             }),
             upload_reports,
@@ -422,7 +437,8 @@ impl Engine {
         self.verification_handler = Some(self.verification_handler(incoming));
         let client = &self.client;
         let events = || self.port.event_sender();
-        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 11] = [
+        let (notices, notice_queue) = mpsc::unbounded_channel();
+        let watchers: [(&'static str, EngineBoxFuture<'static, ()>); 12] = [
             (
                 "upload tracker",
                 Box::pin(watchers::uploads(
@@ -441,18 +457,27 @@ impl Engine {
             ),
             (
                 "session change watcher",
-                Box::pin(watchers::session(
+                Box::pin(session::session(
+                    client.clone(),
                     queues.session_changes,
-                    events(),
+                    queues.persistence,
+                    notices,
                     client
                         .device_id()
                         .expect("authentication was checked during construction")
                         .to_owned(),
-                    self.stop.clone(),
                     self.logouts.clone(),
                 )),
             ),
             (
+                "session notice watcher",
+                Box::pin(session::session_notices(
+                    notice_queue,
+                    events(),
+                    self.stop.clone(),
+                )),
+            ),
+            (
                 "connection state watcher",
                 Box::pin(watchers::connection(
                     self.sync.state(),
@@ -560,7 +585,7 @@ impl Engine {
                         false
                     });
                     if let Some(soft_logout) = rejected {
-                        watchers::reject_session(&events, &stop, soft_logout).await;
+                        session::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))

Added file: src/engine/session.rs+132 additions@@ -0,0 +1,132 @@
+use std::sync::Arc;
+
+use matrix_sdk::{BoxFuture, Client, SessionChange};
+use ruma::OwnedDeviceId;
+use tokio::sync::{
+    broadcast::{self, error::RecvError},
+    mpsc, watch,
+};
+
+use super::Logouts;
+use crate::{Error, Event, Termination, transport::EventSender};
+
+#[cfg(not(target_family = "wasm"))]
+pub(crate) type SaveRecord = dyn Fn(Client) -> BoxFuture<'static, Result<(), Error>> + Send + Sync;
+#[cfg(target_family = "wasm")]
+pub(crate) type SaveRecord = dyn Fn(Client) -> BoxFuture<'static, Result<(), Error>>;
+
+/// how the engine keeps the saved record current, handed over by `Session::build`
+pub(crate) struct Persistence {
+    pub(crate) save: Box<SaveRecord>,
+    /// what the sdk's own save callback made of each refresh, when the store could write
+    /// there and then
+    pub(crate) inline: mpsc::UnboundedReceiver<Result<(), Error>>,
+}
+
+impl Persistence {
+    async fn persist(&mut self, client: &Client) -> Result<(), Error> {
+        let mut inline = None;
+        while let Ok(result) = self.inline.try_recv() {
+            inline = Some(result);
+        }
+        if let Some(Ok(())) = inline {
+            return Ok(());
+        }
+        // an inline save for a newer refresh can land while this one writes, so go again
+        // until what's on disk is what the client holds
+        loop {
+            let tokens = client.session_tokens();
+            (self.save)(client.clone()).await?;
+            if client.session_tokens() == tokens {
+                return Ok(());
+            }
+        }
+    }
+}
+
+pub(super) enum Notice {
+    Event(Event),
+    Rejected(Option<OwnedDeviceId>),
+}
+
+/// saves and notices are split, so a full event stream never holds a save back
+pub(super) async fn session(
+    client: Client,
+    mut changes: broadcast::Receiver<SessionChange>,
+    mut persistence: Option<Persistence>,
+    notices: mpsc::UnboundedSender<Notice>,
+    device_id: OwnedDeviceId,
+    logouts: Arc<watch::Sender<Logouts>>,
+) {
+    loop {
+        let event = match changes.recv().await {
+            // a missed refresh costs one needless save, a missed rejection comes round again
+            Ok(SessionChange::TokensRefreshed) | Err(RecvError::Lagged(_)) => {
+                match persistence.as_mut() {
+                    None => Event::SessionTokensChanged,
+                    Some(persistence) => match persistence.persist(&client).await {
+                        Ok(()) => Event::SessionTokensChanged,
+                        Err(error) => Event::SessionSaveFailed {
+                            message: error.message,
+                        },
+                    },
+                }
+            }
+            Ok(SessionChange::UnknownToken(data)) => {
+                let soft_logout = data.soft_logout.then_some(device_id);
+                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 {
+                    let _ = notices.send(Notice::Rejected(soft_logout));
+                }
+                return;
+            }
+            Err(RecvError::Closed) => return,
+        };
+        if notices.send(Notice::Event(event)).is_err() {
+            return;
+        }
+    }
+}
+
+pub(super) async fn session_notices(
+    mut notices: mpsc::UnboundedReceiver<Notice>,
+    events: EventSender,
+    stop: mpsc::UnboundedSender<Termination>,
+) {
+    while let Some(notice) = notices.recv().await {
+        match notice {
+            Notice::Event(event) => {
+                if events.send(event).await.is_err() {
+                    return;
+                }
+            }
+            Notice::Rejected(soft_logout) => {
+                return reject_session(&events, &stop, soft_logout).await;
+            }
+        }
+    }
+}
+
+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,
+    });
+}

Modified file: src/engine/watchers.rs+8 additions−61 deletions@@ -2,18 +2,18 @@ use std::{collections::HashSet, sync::Arc, time::Duration};
 
 use futures_util::StreamExt;
 use matrix_sdk::{
-    Client, Room, SessionChange,
+    Client, Room,
     ruma::events::{AnyMessageLikeEventContent, room::message::MessageType},
     send_queue::{LocalEchoContent, RoomSendQueueUpdate, SendQueueUpdate},
 };
 use matrix_sdk_ui::sync_service::{State as SyncState, SyncService};
-use ruma::{OwnedDeviceId, OwnedRoomId, OwnedTransactionId, time::Instant};
+use ruma::{OwnedRoomId, OwnedTransactionId, time::Instant};
 use tokio::sync::{broadcast::error::RecvError, mpsc, watch};
 
-use super::{Logouts, executor::TaskSet};
+use super::executor::TaskSet;
 use crate::{
-    ConnectionState, Event, Termination, UploadOutcome, encryption::encryption_state,
-    error::classify_sdk_error, transport::EventSender,
+    ConnectionState, Event, UploadOutcome, encryption::encryption_state, error::classify_sdk_error,
+    transport::EventSender,
 };
 
 const SYNC_RESTART_MIN: Duration = Duration::from_secs(1);
@@ -32,58 +32,6 @@ pub(super) async fn room_subscriptions(
     }
 }
 
-pub(super) async fn session(
-    mut changes: tokio::sync::broadcast::Receiver<SessionChange>,
-    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 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
-            Err(RecvError::Lagged(_)) => Event::SessionTokensChanged,
-            Err(RecvError::Closed) => return,
-        };
-        if events.send(event).await.is_err() {
-            return;
-        }
-    }
-}
-
-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,
@@ -202,10 +150,9 @@ pub(super) async fn room_uploads(
                 (room_id.clone(), transaction_id),
                 UploadOutcome::Sent { event_id },
             ),
-            Ok(RoomSendQueueUpdate::CancelledLocalEvent { transaction_id }) => RoomUpload::Ended(
-                (room_id.clone(), transaction_id),
-                UploadOutcome::Cancelled,
-            ),
+            Ok(RoomSendQueueUpdate::CancelledLocalEvent { transaction_id }) => {
+                RoomUpload::Ended((room_id.clone(), transaction_id), UploadOutcome::Cancelled)
+            }
             Ok(_) => continue,
             Err(RecvError::Lagged(_)) => RoomUpload::Lagged,
             Err(RecvError::Closed) => return,

Modified file: src/model.rs+7 additions−2 deletions@@ -634,9 +634,14 @@ pub enum Event {
     Authenticated {
         user_id: OwnedUserId,
     },
-    /// Refresh tokens changed. Persist a fresh `ClientHandle::session_record()`
-    /// atomically; a record captured before this event may no longer restore.
+    /// the tokens rotated. with a `SessionStore` the record is already saved by now,
+    /// without one persist a fresh `ClientHandle::session_record()` atomically
     SessionTokensChanged,
+    /// the rotated tokens didn't reach the `SessionStore`, so the saved record may not
+    /// restore. the engine tries again on the next rotation
+    SessionSaveFailed {
+        message: String,
+    },
     /// the server stopped honouring our token and the engine stops. a soft logout
     /// names the device to log in again as, keeping the store. otherwise the record
     /// and the store are both dead

Modified file: src/session_store.rs+40 additions−4 deletions@@ -22,6 +22,14 @@ pub trait SessionStore: SendOutsideWasm + SyncOutsideWasm {
     /// replace the saved record, all or nothing
     fn save<'a>(&'a self, record: &'a SessionRecord) -> BoxFuture<'a, Result<(), AccountError>>;
 
+    /// `save`, but without awaiting. the sdk calls it from inside a refresh, before the new
+    /// refresh token is ever used, so a store that can write synchronously closes the gap
+    /// a crash would otherwise lose the login through. `None` leaves the refresh to `save`
+    #[cfg(not(target_family = "wasm"))]
+    fn save_blocking(&self, _record: &SessionRecord) -> Option<Result<(), AccountError>> {
+        None
+    }
+
     /// swap the record for one that only says who to log in again as, keeping the store.
     /// `Session::restore` then answers `Restore::SoftLoggedOut`, even after a restart
     fn soft_log_out(&self) -> BoxFuture<'_, Result<(), AccountError>>;
@@ -113,14 +121,33 @@ impl EncryptedSessionStore {
         sealer: &mut Option<Sealer>,
         record: &SessionRecord,
     ) -> Result<(), AccountError> {
+        let sealed = self.seal(sealer, record)?;
+        backend::write(&self.record, &sealed)
+            .await
+            .map_err(|error| account_error("writing the session record", error))
+    }
+
+    fn seal(
+        &self,
+        sealer: &mut Option<Sealer>,
+        record: &SessionRecord,
+    ) -> Result<Vec<u8>, AccountError> {
         let sealer = match sealer {
             Some(sealer) => sealer,
             empty => empty.insert(Sealer::new(self.store.passphrase())?),
         };
-        let sealed = sealer.seal(record)?;
-        backend::write(&self.record, &sealed)
-            .await
-            .map_err(|error| account_error("writing the session record", error))
+        sealer.seal(record)
+    }
+
+    /// the lock is only ever held around one small write, so losing the race to it just
+    /// hands this refresh to the async save
+    #[cfg(feature = "sqlite")]
+    fn save_now(&self, record: &SessionRecord) -> Option<Result<(), AccountError>> {
+        let mut sealer = self.sealer.try_lock().ok()?;
+        Some(self.seal(&mut sealer, record).and_then(|sealed| {
+            backend::write_blocking(&self.record, &sealed)
+                .map_err(|error| account_error("writing the session record", error))
+        }))
     }
 }
 
@@ -137,6 +164,11 @@ impl SessionStore for EncryptedSessionStore {
         Box::pin(self.save_record(record))
     }
 
+    #[cfg(feature = "sqlite")]
+    fn save_blocking(&self, record: &SessionRecord) -> Option<Result<(), AccountError>> {
+        self.save_now(record)
+    }
+
     fn soft_log_out(&self) -> BoxFuture<'_, Result<(), AccountError>> {
         Box::pin(self.soft_log_out_record())
     }
@@ -211,6 +243,10 @@ mod backend {
     }
 
     pub(super) async fn write(path: &Path, sealed: &[u8]) -> io::Result<()> {
+        write_blocking(path, sealed)
+    }
+
+    pub(super) fn write_blocking(path: &Path, sealed: &[u8]) -> io::Result<()> {
         let mut temporary = path.as_os_str().to_owned();
         temporary.push(".tmp");
         let temporary = PathBuf::from(temporary);

Modified file: src/transport.rs+2 additions−1 deletions@@ -75,7 +75,8 @@ impl Future for Pending {
 }
 
 impl ClientHandle {
-    /// current credentials. persist this atomically on `Event::SessionTokensChanged`, it holds secrets
+    /// current credentials, it holds secrets. the engine saves it for you unless the
+    /// session was built `without_persistence`
     #[cfg(any(feature = "sqlite", feature = "indexeddb"))]
     pub fn session_record(&self) -> Result<crate::SessionRecord, crate::AccountError> {
         crate::account::session_record(&self.client)

Modified file: tests/account.rs+77 additions−9 deletions@@ -1,7 +1,7 @@
 #![cfg(feature = "sqlite")]
 mod support;
 
-use std::{path::PathBuf, time::SystemTime};
+use std::{path::PathBuf, sync::Arc, time::SystemTime};
 
 use macltok::{
     EncryptedSessionStore, EngineConfig, Event, PasswordLogin, Restore, Session, SessionRecord,
@@ -37,7 +37,7 @@ async fn password_session_round_trips_and_discards_with_its_store() {
     let session = Session::login_password(
         PasswordLogin::new(server.uri(), "example", "wordpass")
             .with_device_display_name("macltok test"),
-        &sessions(),
+        Arc::new(sessions()),
     )
     .await
     .unwrap();
@@ -56,8 +56,8 @@ async fn password_session_round_trips_and_discards_with_its_store() {
     );
     assert!(wrong.load().await.is_err());
 
-    let sessions = sessions();
-    let Some(Restore::Resumed(session)) = Session::restore(&sessions).await.unwrap() else {
+    let sessions = Arc::new(sessions());
+    let Some(Restore::Resumed(session)) = Session::restore(sessions.clone()).await.unwrap() else {
         panic!("the saved login didn't resume");
     };
     let (engine, handle, mut events) = session.build(config).await.unwrap();
@@ -72,7 +72,7 @@ async fn password_session_round_trips_and_discards_with_its_store() {
     sessions.discard().await.unwrap();
     assert!(!record_path.exists());
     assert!(!store_path.exists());
-    assert!(Session::restore(&sessions).await.unwrap().is_none());
+    assert!(Session::restore(sessions).await.unwrap().is_none());
     std::fs::remove_dir_all(root).unwrap();
 }
 
@@ -90,18 +90,19 @@ async fn a_restart_after_a_soft_logout_still_knows_the_device() {
     };
     Session::login_password(
         PasswordLogin::new(server.uri(), "example", "wordpass"),
-        &sessions(),
+        Arc::new(sessions()),
     )
     .await
     .unwrap();
     sessions().soft_log_out().await.unwrap();
 
-    let Some(Restore::SoftLoggedOut(soft)) = Session::restore(&sessions()).await.unwrap() else {
+    let Some(Restore::SoftLoggedOut(soft)) = Session::restore(Arc::new(sessions())).await.unwrap()
+    else {
         panic!("a soft logged out record came back as something else");
     };
     // the device matrix-sdk-test's canned login response hands out
     assert_eq!(soft.device_id, "GHTYAJCE");
-    Session::login_password(soft.password_login("wordpass"), &sessions())
+    Session::login_password(soft.password_login("wordpass"), Arc::new(sessions()))
         .await
         .unwrap();
     let requests = server.server().received_requests().await.unwrap();
@@ -113,7 +114,7 @@ async fn a_restart_after_a_soft_logout_still_knows_the_device() {
         .unwrap();
     assert_eq!(relogin["device_id"], "GHTYAJCE");
     assert!(matches!(
-        Session::restore(&sessions()).await.unwrap(),
+        Session::restore(Arc::new(sessions())).await.unwrap(),
         Some(Restore::Resumed(_))
     ));
     std::fs::remove_dir_all(root).unwrap();
@@ -153,6 +154,73 @@ async fn a_torn_or_failed_save_leaves_the_old_record() {
     std::fs::remove_dir_all(root).unwrap();
 }
 
+#[tokio::test]
+async fn a_refresh_is_saved_before_anyone_hears_of_it() {
+    use matrix_sdk_test::test_json;
+    use wiremock::{
+        Mock, ResponseTemplate,
+        matchers::{header, method, path, path_regex},
+    };
+
+    let server = MatrixMockServer::new().await;
+    server.mock_versions().ok().mount().await;
+    server
+        .mock_login()
+        .respond_with(
+            ResponseTemplate::new(200).set_body_json(&*test_json::LOGIN_WITH_REFRESH_TOKEN),
+        )
+        .mount()
+        .await;
+    Mock::given(method("POST"))
+        .and(path_regex("sync"))
+        .and(header("authorization", "Bearer abc123"))
+        .respond_with(ResponseTemplate::new(401).set_body_json(serde_json::json!({
+            "errcode": "M_UNKNOWN_TOKEN",
+            "error": "expired",
+            "soft_logout": false,
+        })))
+        .with_priority(1)
+        .mount(server.server())
+        .await;
+    Mock::given(method("POST"))
+        .and(path("/_matrix/client/v3/refresh"))
+        .respond_with(
+            ResponseTemplate::new(200).set_body_json(&*test_json::REFRESH_TOKEN_WITH_REFRESH_TOKEN),
+        )
+        .mount(server.server())
+        .await;
+    server
+        .mock_sliding_sync()
+        .ok(v5::Response::new("1".to_owned()))
+        .mount()
+        .await;
+    let root = temporary_path();
+    let sessions = Arc::new(EncryptedSessionStore::file(
+        root.join("session"),
+        StoreConfig::sqlite(root.join("store"), "test store passphrase"),
+    ));
+    let session = Session::login_password(
+        PasswordLogin::new(server.uri(), "example", "wordpass"),
+        sessions.clone(),
+    )
+    .await
+    .unwrap();
+    let (engine, handle, mut events) = session
+        .build(EngineConfig::default().with_initial_history_limit(0))
+        .await
+        .unwrap();
+    let running = tokio::spawn(engine.run());
+    support::receive_until(&mut events, |event| {
+        matches!(event, Event::SessionTokensChanged)
+    })
+    .await;
+    let saved = serde_json::to_value(sessions.load().await.unwrap().unwrap()).unwrap();
+    assert_eq!(saved["authentication"]["session"]["refresh_token"], "wxyz");
+    support::shut_down(&handle, &mut events, running).await;
+    drop(handle);
+    std::fs::remove_dir_all(root).unwrap();
+}
+
 fn record(homeserver: &str) -> SessionRecord {
     serde_json::from_value(serde_json::json!({
         "homeserver": homeserver,

macltok

Commit f12377bce9b9

Commit `f12377bce9b9`