Commit 698e2238368e
tarball1 file changed+69-10
@@ -1,7 +1,7 @@
11
use std::{22
collections::HashMap,33
path::{Path, PathBuf},4
- sync::Arc,4
+ sync::{Arc, PoisonError, Weak},55
time::Duration,66
};77
@@ -13,7 +13,11 @@ use macltok::{
1313
};1414
use rand_core::{OsRng, RngCore};1515
use sha2::{Digest, Sha256};16
-use tokio::{sync::Mutex, task::JoinHandle, time::timeout};16
+use tokio::{17
+ sync::{Mutex, OwnedMutexGuard},18
+ task::JoinHandle,19
+ time::timeout,20
+};1721
1822
use malice_core::{1923
error::{CoreError, Failure},@@ -36,9 +40,10 @@ pub struct UserSession {
3640
3741
struct Inner {3842
sessions: tokio::sync::RwLock<HashMap<String, Arc<UserSession>>>,39
- /// Serialises concurrent restores of the same sid so two engines never40
- /// open the same SQLite store at once.41
- init_lock: Mutex<()>,43
+ /// Serialises restores, logins and removals of the same sid so two engines44
+ /// never open the same SQLite store at once, without one slow homeserver45
+ /// holding up every other user.46
+ sid_locks: std::sync::Mutex<HashMap<String, Weak<Mutex<()>>>>,4247
data_dir: PathBuf,4348
master_key: [u8; 32],4449
cookie_secure: bool,@@ -72,7 +77,7 @@ impl AppState {
7277
Self {7378
inner: Arc::new(Inner {7479
sessions: tokio::sync::RwLock::new(HashMap::new()),75
- init_lock: Mutex::new(()),80
+ sid_locks: std::sync::Mutex::new(HashMap::new()),7681
data_dir,7782
master_key,7883
cookie_secure,@@ -89,6 +94,25 @@ impl AppState {
8994
self.inner.cookie_secure9095
}9196
97
+ async fn lock_sid(&self, sid: &str) -> OwnedMutexGuard<()> {98
+ let lock = {99
+ let mut locks = self100
+ .inner101
+ .sid_locks102
+ .lock()103
+ .unwrap_or_else(PoisonError::into_inner);104
+ locks.retain(|_, lock| lock.strong_count() > 0);105
+ if let Some(lock) = locks.get(sid).and_then(Weak::upgrade) {106
+ lock107
+ } else {108
+ let lock = Arc::new(Mutex::new(()));109
+ locks.insert(sid.to_owned(), Arc::downgrade(&lock));110
+ lock111
+ }112
+ };113
+ lock.lock_owned().await114
+ }115
+92116
fn sessions_dir(&self) -> PathBuf {93117
self.inner.data_dir.join("sessions")94118
}@@ -144,7 +168,7 @@ impl AppState {
144168
if let Some(user) = self.inner.sessions.read().await.get(sid).cloned() {145169
return Ok(user);146170
}147
- let _init = self.inner.init_lock.lock().await;171
+ let _init = self.lock_sid(sid).await;148172
if let Some(user) = self.inner.sessions.read().await.get(sid).cloned() {149173
return Ok(user);150174
}@@ -200,7 +224,7 @@ impl AppState {
200224
sessions.clone(),201225
)202226
.await?;203
- let _init = self.inner.init_lock.lock().await;227
+ let _init = self.lock_sid(sid).await;204228
return self205229
.start_user_session(sid.to_owned(), session, sessions)206230
.await@@ -212,7 +236,7 @@ impl AppState {
212236
let sid = new_sid();213237
let sessions = self.session_store(&sid);214238
let session = login_with(login, &password, sessions.clone()).await?;215
- let _init = self.inner.init_lock.lock().await;239
+ let _init = self.lock_sid(&sid).await;216240
self.start_user_session(sid, session, sessions)217241
.await218242
.map_err(|error| CoreError::new(Failure::Upstream, error))@@ -282,7 +306,7 @@ impl AppState {
282306
pub async fn remove_session(&self, sid: &str, retain: Retain) {283307
// A restore must not reopen the store between removing the map entry284308
// and deleting its persisted credentials.285
- let _init = self.inner.init_lock.lock().await;309
+ let _init = self.lock_sid(sid).await;286310
let user = self.inner.sessions.write().await.remove(sid);287311
let sessions = match &user {288312
Some(user) => user.sessions.clone(),@@ -533,3 +557,38 @@ fn restrict_file(path: &Path) {
533557
534558
#[cfg(not(unix))]535559
fn restrict_file(_path: &Path) {}560
+561
+#[cfg(test)]562
+mod tests {563
+ use super::*;564
+565
+ fn state() -> (AppState, tempfile::TempDir) {566
+ let dir = tempfile::tempdir().expect("temp dir");567
+ (568
+ AppState::new(dir.path().to_owned(), [7; 32], false, false),569
+ dir,570
+ )571
+ }572
+573
+ #[tokio::test]574
+ async fn sid_locks_only_serialise_the_same_sid() {575
+ let (state, _dir) = state();576
+ let held = state.lock_sid("alpha").await;577
+ let other = timeout(Duration::from_millis(50), state.lock_sid("beta")).await;578
+ assert!(other.is_ok(), "a different sid must not wait");579
+ let same = timeout(Duration::from_millis(50), state.lock_sid("alpha")).await;580
+ assert!(same.is_err(), "the same sid must wait for the holder");581
+ drop(held);582
+ let again = timeout(Duration::from_millis(50), state.lock_sid("alpha")).await;583
+ assert!(again.is_ok(), "the sid is free once the holder drops");584
+ }585
+586
+ #[tokio::test]587
+ async fn sid_locks_forget_released_sids() {588
+ let (state, _dir) = state();589
+ drop(state.lock_sid("alpha").await);590
+ drop(state.lock_sid("beta").await);591
+ let live = state.inner.sid_locks.lock().unwrap().len();592
+ assert!(live <= 1, "released locks are pruned on the next lookup");593
+ }594
+}