macltok

public

Star0

Fork 0

Watch0

25 commits 3 branches 0 tags 0 releases3.0 MiB

Commit `6db49c0ecf71`

tarball

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

6 files changed+268-89

session: make record writes safe and let IndexedDB deletes report blocked

Modified file: Cargo.lock+4 additions@@ -1828,6 +1828,7 @@ dependencies = [
  "getrandom 0.4.3",
  "http",
  "imbl",
+ "js-sys",
  "matrix-sdk",
  "matrix-sdk-crypto",
  "matrix-sdk-store-encryption",
@@ -1845,6 +1846,9 @@ dependencies = [
  "tracing",
  "url",
  "uuid",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
  "wiremock",
 ]
 

Modified file: Cargo.toml+14 additions@@ -16,6 +16,10 @@ indexeddb = [
     "dep:url",
     "dep:matrix-sdk-store-encryption",
     "dep:indexed_db_futures",
+    "dep:js-sys",
+    "dep:wasm-bindgen",
+    "dep:wasm-bindgen-futures",
+    "dep:web-sys",
 ]
 
 [dependencies]
@@ -40,11 +44,21 @@ getrandom = { version = "0.2", features = ["js"] }
 getrandom3 = { package = "getrandom", version = "0.3", features = ["wasm_js"] }
 getrandom4 = { package = "getrandom", version = "0.4", features = ["wasm_js"] }
 indexed_db_futures = { package = "matrix_indexed_db_futures", version = "0.7.0", optional = true, features = ["typed-arrays"] }
+js-sys = { version = "0.3.105", optional = true }
 matrix-sdk = { version = "0.18.0", default-features = false, features = ["js"] }
 matrix-sdk-ui = { version = "0.18.0", default-features = false, features = ["js"] }
 # the sdk's own wasm reqwest leaves out stream, which the bounded media download needs
 reqwest = { version = "0.13.1", default-features = false, features = ["stream"] }
 uuid = { version = "1", features = ["js"] }
+wasm-bindgen = { version = "0.2.128", optional = true }
+wasm-bindgen-futures = { version = "0.4.78", optional = true }
+web-sys = { version = "0.3.105", optional = true, features = [
+    "IdbFactory",
+    "IdbOpenDbRequest",
+    "IdbRequest",
+    "Window",
+    "WorkerGlobalScope",
+] }
 
 [dev-dependencies]
 matrix-sdk = { version = "0.18.0", default-features = false, features = ["testing"] }

Modified file: README.md+3 additions@@ -128,6 +128,9 @@ what's different in a browser
   restarts itself with backoff, that part is ours
 - two tabs or workers on one store each need `StoreConfig::with_lock_holder`
   with a name of their own, or they both think they hold the crypto store
+- `discard` while another tab still has the store open answers
+  `StoreDeletePending`. the delete lands once that tab lets go, and reopening
+  the store waits behind it
 
 ## Dependency policy
 

Modified file: src/account.rs+5 additions−1 deletions@@ -245,7 +245,7 @@ impl StoreConfig {
             #[cfg(feature = "indexeddb")]
             StoreLocation::IndexedDb(name) => crate::session_store::delete_indexeddb_store(name)
                 .await
-                .map_err(|error| account_error("deleting the Matrix store", error)),
+                .map_err(|error| error.into_account_error("deleting the Matrix store")),
         }
     }
 
@@ -348,6 +348,10 @@ pub enum AccountError {
     UnsupportedAuthentication,
     #[error("the OAuth callback was already submitted")]
     OAuthCallbackAlreadySubmitted,
+    /// an IndexedDB delete is waiting on the store's other connections. it goes through
+    /// once they close, and opening the store again waits behind it
+    #[error("the store is still open elsewhere, it will be deleted once that closes")]
+    StoreDeletePending,
     #[error(transparent)]
     Engine(#[from] EngineError),
 }

Modified file: src/session_store.rs+236 additions−85 deletions@@ -3,7 +3,6 @@ use std::path::{Path, PathBuf};
 
 use matrix_sdk::{BoxFuture, SendOutsideWasm, SyncOutsideWasm};
 use matrix_sdk_store_encryption::StoreCipher;
-use tokio::sync::Mutex;
 
 use crate::{AccountError, SessionRecord, StoreConfig, account::account_error};
 
@@ -12,6 +11,11 @@ type RecordLocation = PathBuf;
 #[cfg(feature = "indexeddb")]
 type RecordLocation = String;
 
+#[cfg(not(target_family = "wasm"))]
+type Lock<T> = std::sync::Mutex<T>;
+#[cfg(target_family = "wasm")]
+type Lock<T> = tokio::sync::Mutex<T>;
+
 /// one login's record and the matrix store it belongs to. the engine rotates refresh tokens
 /// and each works once, so a `save` has to land before the next refresh or the login is gone
 pub trait SessionStore: SendOutsideWasm + SyncOutsideWasm {
@@ -43,7 +47,7 @@ pub trait SessionStore: SendOutsideWasm + SyncOutsideWasm {
 pub struct EncryptedSessionStore {
     record: RecordLocation,
     store: StoreConfig,
-    sealer: Mutex<Option<Sealer>>,
+    sealer: Lock<Option<Sealer>>,
 }
 
 struct Sealer {
@@ -70,50 +74,126 @@ impl EncryptedSessionStore {
         Self {
             record,
             store,
-            sealer: Mutex::new(None),
+            sealer: Lock::new(None),
         }
     }
 
-    async fn load_record(&self) -> Result<Option<SessionRecord>, AccountError> {
-        let mut sealer = self.sealer.lock().await;
-        self.read(&mut sealer).await
+    fn open(
+        &self,
+        sealer: &mut Option<Sealer>,
+        sealed: Option<Vec<u8>>,
+    ) -> Result<Option<SessionRecord>, AccountError> {
+        let Some(sealed) = sealed else {
+            return Ok(None);
+        };
+        let (opened, record) = unseal(self.store.passphrase(), &sealed)?;
+        *sealer = Some(opened);
+        Ok(Some(record))
     }
 
-    async fn save_record(&self, record: &SessionRecord) -> Result<(), AccountError> {
-        let mut sealer = self.sealer.lock().await;
-        self.write(&mut sealer, record).await
+    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())?),
+        };
+        sealer.seal(record)
     }
+}
 
-    async fn soft_log_out_record(&self) -> Result<(), AccountError> {
-        let mut sealer = self.sealer.lock().await;
-        match self.read(&mut sealer).await? {
-            Some(record) => self.write(&mut sealer, &record.soft_logged_out()).await,
+#[cfg(not(target_family = "wasm"))]
+impl EncryptedSessionStore {
+    fn lock(&self) -> std::sync::MutexGuard<'_, Option<Sealer>> {
+        self.sealer
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+    }
+
+    fn read(&self, sealer: &mut Option<Sealer>) -> Result<Option<SessionRecord>, AccountError> {
+        let sealed = backend::read(&self.record)
+            .map_err(|error| account_error("reading the session record", error))?;
+        self.open(sealer, sealed)
+    }
+
+    fn write(
+        &self,
+        sealer: &mut Option<Sealer>,
+        record: &SessionRecord,
+    ) -> Result<(), AccountError> {
+        let sealed = self.seal(sealer, record)?;
+        backend::write(&self.record, &sealed)
+            .map_err(|error| account_error("writing the session record", error))
+    }
+
+    fn load_record(&self) -> Result<Option<SessionRecord>, AccountError> {
+        self.read(&mut self.lock())
+    }
+
+    fn save_record(&self, record: &SessionRecord) -> Result<(), AccountError> {
+        self.write(&mut self.lock(), record)
+    }
+
+    fn soft_log_out_record(&self) -> Result<(), AccountError> {
+        let mut sealer = self.lock();
+        match self.read(&mut sealer)? {
+            Some(record) => self.write(&mut sealer, &record.soft_logged_out()),
             None => Ok(()),
         }
     }
 
-    async fn discard_all(&self) -> Result<(), AccountError> {
-        let mut sealer = self.sealer.lock().await;
+    fn forget_record(&self) -> Result<(), AccountError> {
+        let mut sealer = self.lock();
         backend::remove(&self.record)
-            .await
             .map_err(|error| account_error("removing the session record", error))?;
         *sealer = None;
-        self.store.delete().await
+        Ok(())
+    }
+}
+
+#[cfg(not(target_family = "wasm"))]
+impl SessionStore for EncryptedSessionStore {
+    fn store(&self) -> &StoreConfig {
+        &self.store
     }
 
+    fn load(&self) -> BoxFuture<'_, Result<Option<SessionRecord>, AccountError>> {
+        Box::pin(async move { self.load_record() })
+    }
+
+    fn save<'a>(&'a self, record: &'a SessionRecord) -> BoxFuture<'a, Result<(), AccountError>> {
+        Box::pin(async move { self.save_record(record) })
+    }
+
+    /// blocks the calling worker for the write and its fsyncs, about a millisecond
+    fn save_blocking(&self, record: &SessionRecord) -> Option<Result<(), AccountError>> {
+        Some(self.save_record(record))
+    }
+
+    fn soft_log_out(&self) -> BoxFuture<'_, Result<(), AccountError>> {
+        Box::pin(async move { self.soft_log_out_record() })
+    }
+
+    fn discard(&self) -> BoxFuture<'_, Result<(), AccountError>> {
+        Box::pin(async move {
+            self.forget_record()?;
+            self.store.delete().await
+        })
+    }
+}
+
+#[cfg(target_family = "wasm")]
+impl EncryptedSessionStore {
     async fn read(
         &self,
         sealer: &mut Option<Sealer>,
     ) -> Result<Option<SessionRecord>, AccountError> {
-        let Some(sealed) = backend::read(&self.record)
+        let sealed = backend::read(&self.record)
             .await
-            .map_err(|error| account_error("reading the session record", error))?
-        else {
-            return Ok(None);
-        };
-        let (opened, record) = unseal(self.store.passphrase(), &sealed)?;
-        *sealer = Some(opened);
-        Ok(Some(record))
+            .map_err(|error| account_error("reading the session record", error))?;
+        self.open(sealer, sealed)
     }
 
     async fn write(
@@ -127,30 +207,33 @@ impl EncryptedSessionStore {
             .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())?),
-        };
-        sealer.seal(record)
+    async fn load_record(&self) -> Result<Option<SessionRecord>, AccountError> {
+        self.read(&mut *self.sealer.lock().await).await
     }
 
-    /// 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))
-        }))
+    async fn save_record(&self, record: &SessionRecord) -> Result<(), AccountError> {
+        self.write(&mut *self.sealer.lock().await, record).await
+    }
+
+    async fn soft_log_out_record(&self) -> Result<(), AccountError> {
+        let mut sealer = self.sealer.lock().await;
+        match self.read(&mut sealer).await? {
+            Some(record) => self.write(&mut sealer, &record.soft_logged_out()).await,
+            None => Ok(()),
+        }
+    }
+
+    async fn discard_all(&self) -> Result<(), AccountError> {
+        let mut sealer = self.sealer.lock().await;
+        backend::remove(&self.record)
+            .await
+            .map_err(|error| error.into_account_error("removing the session record"))?;
+        *sealer = None;
+        self.store.delete().await
     }
 }
 
+#[cfg(target_family = "wasm")]
 impl SessionStore for EncryptedSessionStore {
     fn store(&self) -> &StoreConfig {
         &self.store
@@ -164,11 +247,6 @@ 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())
     }
@@ -229,12 +307,16 @@ fn unseal(passphrase: &str, sealed: &[u8]) -> Result<(Sealer, SessionRecord), Ac
 #[cfg(feature = "sqlite")]
 mod backend {
     use std::{
+        ffi::OsString,
         fs::{self, File, OpenOptions},
         io::{self, ErrorKind, Write},
         path::{Path, PathBuf},
+        sync::atomic::{AtomicU64, Ordering},
     };
 
-    pub(super) async fn read(path: &Path) -> io::Result<Option<Vec<u8>>> {
+    static WRITES: AtomicU64 = AtomicU64::new(0);
+
+    pub(super) fn read(path: &Path) -> io::Result<Option<Vec<u8>>> {
         match fs::read(path) {
             Ok(sealed) => Ok(Some(sealed)),
             Err(error) if error.kind() == ErrorKind::NotFound => Ok(None),
@@ -242,59 +324,68 @@ 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);
+    pub(super) fn write(path: &Path, sealed: &[u8]) -> io::Result<()> {
         let parent = match path.parent() {
             Some(parent) if !parent.as_os_str().is_empty() => parent,
             _ => Path::new("."),
         };
         fs::create_dir_all(parent)?;
-        let written = replace(path, &temporary, parent, sealed);
+        let (temporary, mut file) = temporary_file(path)?;
+        let written = replace(path, &temporary, &mut file, parent, sealed);
         if written.is_err() {
             let _ = fs::remove_file(&temporary);
         }
         written
     }
 
-    fn replace(path: &Path, temporary: &Path, parent: &Path, sealed: &[u8]) -> io::Result<()> {
-        let mut file = private_file(temporary)?;
+    fn temporary_file(path: &Path) -> io::Result<(PathBuf, File)> {
+        loop {
+            let mut name = OsString::from(path.as_os_str());
+            let write = WRITES.fetch_add(1, Ordering::Relaxed);
+            name.push(format!(".{}.{write}.tmp", std::process::id()));
+            let temporary = PathBuf::from(name);
+            match private_file(&temporary) {
+                Ok(file) => return Ok((temporary, file)),
+                Err(error) if error.kind() == ErrorKind::AlreadyExists => {}
+                Err(error) => return Err(error),
+            }
+        }
+    }
+
+    fn replace(
+        path: &Path,
+        temporary: &Path,
+        file: &mut File,
+        parent: &Path,
+        sealed: &[u8],
+    ) -> io::Result<()> {
         file.write_all(sealed)?;
         file.sync_all()?;
         fs::rename(temporary, path)?;
         sync_directory(parent)
     }
 
-    pub(super) async fn remove(path: &Path) -> io::Result<()> {
+    pub(super) fn remove(path: &Path) -> io::Result<()> {
         match fs::remove_file(path) {
             Err(error) if error.kind() != ErrorKind::NotFound => Err(error),
             _ => Ok(()),
         }
     }
 
+    /// `create_new` is `O_EXCL`, which posix says never follows a symlink, even a dangling one
     #[cfg(unix)]
     fn private_file(path: &Path) -> io::Result<File> {
         use std::os::unix::fs::OpenOptionsExt;
         OpenOptions::new()
             .write(true)
-            .create(true)
-            .truncate(true)
+            .create_new(true)
             .mode(0o600)
             .open(path)
     }
 
     #[cfg(not(unix))]
     fn private_file(path: &Path) -> io::Result<File> {
-        OpenOptions::new()
-            .write(true)
-            .create(true)
-            .truncate(true)
-            .open(path)
+        OpenOptions::new().write(true).create_new(true).open(path)
     }
 
     /// a rename isn't durable until its directory is synced
@@ -334,20 +425,36 @@ mod backend {
             .await
     }
 
+    /// the crate's `Database` never closes itself on drop, and an open one blocks deleting it
     pub(super) async fn read(name: &str) -> Result<Option<Vec<u8>>, OpenDbError> {
         let database = open(name).await?;
-        let sealed: Option<Uint8Array> = database
+        let sealed = get(&database).await;
+        database.close();
+        Ok(sealed?.map(Into::into))
+    }
+
+    pub(super) async fn write(name: &str, sealed: &[u8]) -> Result<(), OpenDbError> {
+        let database = open(name).await?;
+        let written = put(&database, sealed).await;
+        database.close();
+        Ok(written?)
+    }
+
+    pub(super) async fn remove(name: &str) -> Result<(), super::DeleteError> {
+        super::delete_database(name).await
+    }
+
+    async fn get(database: &Database) -> Result<Option<Uint8Array>, Error> {
+        database
             .transaction(RECORDS)
             .with_mode(TransactionMode::Readonly)
             .build()?
             .object_store(RECORDS)?
             .get(KEY)
-            .await?;
-        Ok(sealed.map(Into::into))
+            .await
     }
 
-    pub(super) async fn write(name: &str, sealed: &[u8]) -> Result<(), OpenDbError> {
-        let database = open(name).await?;
+    async fn put(database: &Database, sealed: &[u8]) -> Result<(), Error> {
         let transaction = database
             .transaction(RECORDS)
             .with_mode(TransactionMode::Readwrite)
@@ -357,11 +464,7 @@ mod backend {
             .put(Uint8ArraySlice::new(sealed))
             .with_key(KEY)
             .build()?;
-        Ok(transaction.commit().await?)
-    }
-
-    pub(super) async fn remove(name: &str) -> Result<(), OpenDbError> {
-        Ok(Database::delete_by_name(name)?.await?)
+        transaction.commit().await
     }
 }
 
@@ -391,11 +494,7 @@ pub(crate) fn delete_sqlite_store(path: &Path) -> std::io::Result<()> {
 }
 
 #[cfg(feature = "indexeddb")]
-pub(crate) async fn delete_indexeddb_store(
-    name: &str,
-) -> Result<(), indexed_db_futures::error::OpenDbError> {
-    use indexed_db_futures::database::Database;
-
+pub(crate) async fn delete_indexeddb_store(name: &str) -> Result<(), DeleteError> {
     // the databases matrix-sdk-indexeddb 0.18 opens under one store name
     for suffix in [
         "",
@@ -405,7 +504,59 @@ pub(crate) async fn delete_indexeddb_store(
         "::event_cache",
         "::media",
     ] {
-        Database::delete_by_name(&format!("{name}{suffix}"))?.await?;
+        delete_database(&format!("{name}{suffix}")).await?;
     }
     Ok(())
 }
+
+#[cfg(feature = "indexeddb")]
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum DeleteError {
+    #[error("the database is still open somewhere, it goes once that closes")]
+    Blocked,
+    #[error("this global scope has no IndexedDB")]
+    Unavailable,
+    #[error("deleting the database failed, {0}")]
+    Failed(String),
+}
+
+#[cfg(feature = "indexeddb")]
+impl DeleteError {
+    pub(crate) fn into_account_error(self, context: &'static str) -> AccountError {
+        match self {
+            Self::Blocked => AccountError::StoreDeletePending,
+            error => account_error(context, error),
+        }
+    }
+}
+
+/// matrix_indexed_db_futures 0.7 ignores `blocked` on a delete, which then waits forever
+#[cfg(feature = "indexeddb")]
+async fn delete_database(name: &str) -> Result<(), DeleteError> {
+    use wasm_bindgen::{JsCast, JsValue};
+
+    const BLOCKED: &str = "blocked";
+    let global = js_sys::global();
+    let factory = if let Some(window) = global.dyn_ref::<web_sys::Window>() {
+        window.indexed_db()
+    } else if let Some(worker) = global.dyn_ref::<web_sys::WorkerGlobalScope>() {
+        worker.indexed_db()
+    } else {
+        return Err(DeleteError::Unavailable);
+    };
+    let factory = factory.ok().flatten().ok_or(DeleteError::Unavailable)?;
+    let request = factory
+        .delete_database(name)
+        .map_err(|error| DeleteError::Failed(format!("{error:?}")))?;
+    let settled = js_sys::Promise::new(&mut |resolve, reject| {
+        request.set_onsuccess(Some(&resolve));
+        request.set_onerror(Some(&reject));
+        let blocked = reject.bind1(&JsValue::NULL, &JsValue::from_str(BLOCKED));
+        request.set_onblocked(Some(blocked.unchecked_ref()));
+    });
+    match wasm_bindgen_futures::JsFuture::from(settled).await {
+        Ok(_) => Ok(()),
+        Err(reason) if reason.as_string().as_deref() == Some(BLOCKED) => Err(DeleteError::Blocked),
+        Err(_) => Err(DeleteError::Failed(format!("{:?}", request.error()))),
+    }
+}

Modified file: tests/account.rs+6 additions−3 deletions@@ -130,7 +130,7 @@ async fn a_torn_or_failed_save_leaves_the_old_record() {
             StoreConfig::sqlite(root.join("store"), "test store passphrase"),
         )
     };
-    let temporary = root.join("session.tmp");
+    let temporary = root.join("session.1.0.tmp");
     sessions()
         .save(&record("https://old.example"))
         .await
@@ -140,8 +140,10 @@ async fn a_torn_or_failed_save_leaves_the_old_record() {
     let loaded = sessions().load().await.unwrap().unwrap();
     assert_eq!(loaded.homeserver(), "https://old.example");
 
-    std::fs::remove_file(&temporary).unwrap();
-    std::fs::create_dir(&temporary).unwrap();
+    let writable = std::fs::metadata(&root).unwrap().permissions();
+    let mut read_only = writable.clone();
+    read_only.set_readonly(true);
+    std::fs::set_permissions(&root, read_only).unwrap();
     assert!(
         sessions()
             .save(&record("https://new.example"))
@@ -151,6 +153,7 @@ async fn a_torn_or_failed_save_leaves_the_old_record() {
     let loaded = sessions().load().await.unwrap().unwrap();
     assert_eq!(loaded.homeserver(), "https://old.example");
 
+    std::fs::set_permissions(&root, writable).unwrap();
     std::fs::remove_dir_all(root).unwrap();
 }

macltok

Commit 6db49c0ecf71

Commit `6db49c0ecf71`