continue!

Hywan · Hywan · commit 891db1461803 · 2025-09-26T12:02:10.000+02:00
diff --git a/crates/matrix-sdk-common/src/cross_process_lock.rs b/crates/matrix-sdk-common/src/cross_process_lock.rs
@@ -238,21 +238,15 @@ where
     }
 
     /// Try to lock once, returns whether the lock was obtained or not.
+    ///
+    /// The lock can be obtained but it can be dirty. In all cases, the renew
+    /// task will run in the background.
     #[instrument(skip(self), fields(?self.lock_key, ?self.lock_holder))]
-    pub async fn try_lock_once(
-        &self,
-    ) -> Result<Option<CrossProcessLockGuard>, CrossProcessLockError> {
+    pub async fn try_lock_once(&self) -> Result<CrossProcessLockResult, CrossProcessLockError> {
         // Hold onto the locking attempt mutex for the entire lifetime of this
         // function, to avoid multiple reentrant calls.
         let mut _attempt = self.locking_attempt.lock().await;
 
-        // The lock is already dirtied? Let's stop here.
-        if self.is_dirty() {
-            return Err(CrossProcessLockError::LockDirtied(CrossProcessLockGuard::new(
-                self.num_holders.clone(),
-            )));
-        }
-
         // If another thread obtained the lock, make sure to only superficially increase
         // the number of holders, and carry on.
         if self.num_holders.load(Ordering::SeqCst) > 0 {
@@ -261,9 +255,12 @@ where
             // was taken by at least one thread, and after this call it will be
             // taken by at least one thread.
             trace!("We already had the lock, incrementing holder count");
+
             self.num_holders.fetch_add(1, Ordering::SeqCst);
-            let guard = CrossProcessLockGuard { num_holders: self.num_holders.clone() };
-            return Ok(Some(guard));
+
+            return Ok(CrossProcessLockResult::Clean(CrossProcessLockGuard::new(
+                self.num_holders.clone(),
+            )));
         }
 
         if let Some(new_generation) = self
@@ -287,10 +284,6 @@ where
                         "The lock has been acquired, but it's been dirtied!"
                     );
                     self.is_dirty.store(true, Ordering::SeqCst);
-
-                    return Err(CrossProcessLockError::LockDirtied(CrossProcessLockGuard::new(
-                        self.num_holders.clone(),
-                    )));
                 }
 
                 // This was the same generation, no problem.
@@ -302,7 +295,7 @@ where
             trace!("Lock acquired!");
         } else {
             trace!("Couldn't acquire the lock immediately.");
-            return Ok(None);
+            return Ok(CrossProcessLockResult::Unobtained);
         }
 
         trace!("Acquired the lock, spawning the lease extension task.");
@@ -357,42 +350,22 @@ where
                     }
                 }
 
-                // The lock has been dirtied. Exit the loop.
-                if this.is_dirty() {
-                    break;
-                }
-
                 sleep(Duration::from_millis(EXTEND_LEASE_EVERY_MS)).await;
 
                 match this
                     .locker
                     .try_lock(LEASE_DURATION_MS, &this.lock_key, &this.lock_holder)
                     .await
                 {
-                    Ok(Some(new_generation)) => {
-                        match this.lock_generation.swap(new_generation, Ordering::SeqCst) {
-                            // It's impossible to renew the lock if the lock wasn't acquired at
-                            // least once. This is unreachable.
-                            NO_CROSS_PROCESS_LOCK_GENERATION => unreachable!(
-                                "It's impossible to renew a lock lease that has not been acquired once"
-                            ),
-
-                            // This was NOT the same generation, the lock has been dirtied!
-                            previous_generation if previous_generation != new_generation => {
-                                warn!(
-                                    ?previous_generation,
-                                    ?new_generation,
-                                    "The lock lease has been renewed, but it's been dirtied!"
-                                );
-                                this.is_dirty.store(true, Ordering::SeqCst);
-
-                                // Exit the loop.
-                                break;
-                            }
-
-                            // This was the same generation, no problem.
-                            _ => {}
-                        }
+                    Ok(Some(_generation)) => {
+                        // It's impossible that the generation can be different
+                        // from the previous generation.
+                        //
+                        // As long as the task runs, the lock is renewed, so the
+                        // generation remains the same. If the lock is not
+                        // taken, it's because the lease has expired, which is
+                        // represented by the `Ok(None)` value, and the task
+                        // must stop.
                     }
 
                     Ok(None) => {
@@ -414,7 +387,13 @@ where
 
         self.num_holders.fetch_add(1, Ordering::SeqCst);
 
-        Ok(Some(CrossProcessLockGuard::new(self.num_holders.clone())))
+        let guard = CrossProcessLockGuard::new(self.num_holders.clone());
+
+        Ok(if self.is_dirty() {
+            CrossProcessLockResult::Dirty(guard)
+        } else {
+            CrossProcessLockResult::Clean(guard)
+        })
     }
 
     /// Attempt to take the lock, with exponential backoff if the lock has
@@ -436,7 +415,7 @@ where
         // lock in `try_lock_once` should sequentialize it all.
 
         loop {
-            if let Some(guard) = self.try_lock_once().await? {
+            if let Some(guard) = self.try_lock_once().await?.ok() {
                 // Reset backoff before returning, for the next attempt to lock.
                 *self.backoff.lock().await = WaitingTime::Some(INITIAL_BACKOFF_MS);
                 return Ok(guard);
@@ -473,18 +452,53 @@ where
     }
 }
 
+/// Represent the result of a locking attempt, either by
+/// [`CrossProcessLock::try_lock_once`] or [`CrossProcessLock::spin_lock`].
+#[derive(Debug)]
+pub enum CrossProcessLockResult {
+    /// The lock has been obtained successfully, all good.
+    Clean(CrossProcessLockGuard),
+
+    /// The lock has been obtained successfully, but the lock is dirty!
+    ///
+    /// This holder has obtained this cross-process lock once, then another
+    /// holder has obtained this cross-process lock _before_ this holder
+    /// obtained it again. The lock is marked as dirty. It means the value
+    /// protected by the cross-process lock may need to be reloaded if
+    /// synchronisation is important.
+    Dirty(CrossProcessLockGuard),
+
+    /// The lock has not been obtained.
+    Unobtained,
+}
+
+impl CrossProcessLockResult {
+    pub fn is_dirty(&self) -> bool {
+        matches!(self, Self::Dirty(_))
+    }
+
+    /// Convert from [`CrossProcessLockResult`] to
+    /// [`Option<T>`] where `T` is [`CrossProcessLockGuard`].
+    pub fn ok(self) -> Option<CrossProcessLockGuard> {
+        match self {
+            Self::Clean(guard) | Self::Dirty(guard) => Some(guard),
+            Self::Unobtained => None,
+        }
+    }
+
+    /// Return `true` if the lock has been obtained, `false` otherwise.
+    pub fn is_ok(&self) -> bool {
+        matches!(self, Self::Clean(_) | Self::Dirty(_))
+    }
+}
+
 /// Error related to the locking API of the store.
 #[derive(Debug, thiserror::Error)]
 pub enum CrossProcessLockError {
     /// Spent too long waiting for a database lock.
     #[error("a lock timed out")]
     LockTimeout,
 
-    /// The lock has been dirtied, i.e. it means another process took the lock
-    /// while this process was holding it.
-    #[error("a lock has been dirtied")]
-    LockDirtied(CrossProcessLockGuard),
-
     #[error(transparent)]
     #[cfg(not(target_family = "wasm"))]
     TryLockError(#[from] Box<dyn Error + Send + Sync>),
@@ -499,6 +513,7 @@ pub enum CrossProcessLockError {
 mod tests {
     use std::{
         collections::HashMap,
+        ops::Not,
         sync::{Arc, RwLock, atomic},
     };
 
@@ -510,8 +525,8 @@ mod tests {
     };
 
     use super::{
-        CrossProcessLock, CrossProcessLockError, CrossProcessLockGeneration, CrossProcessLockGuard,
-        EXTEND_LEASE_EVERY_MS, TryLock,
+        CrossProcessLock, CrossProcessLockError, CrossProcessLockGeneration,
+        CrossProcessLockResult, EXTEND_LEASE_EVERY_MS, TryLock,
         memory_store_helper::{Lease, try_take_leased_lock},
     };
 
@@ -548,8 +563,8 @@ mod tests {
         }
     }
 
-    async fn release_lock(guard: Option<CrossProcessLockGuard>) {
-        drop(guard);
+    async fn release_lock(result: CrossProcessLockResult) {
+        drop(result);
         sleep(Duration::from_millis(EXTEND_LEASE_EVERY_MS)).await;
     }
 
@@ -561,19 +576,19 @@ mod tests {
         let lock = CrossProcessLock::new(store, "key".to_owned(), "first".to_owned());
 
         // The lock plain works when used with a single holder.
-        let acquired = lock.try_lock_once().await?;
-        assert!(acquired.is_some());
+        let result = lock.try_lock_once().await?;
+        assert!(result.is_ok());
         assert_eq!(lock.num_holders.load(atomic::Ordering::SeqCst), 1);
 
         // Releasing works.
-        release_lock(acquired).await;
+        release_lock(result).await;
         assert_eq!(lock.num_holders.load(atomic::Ordering::SeqCst), 0);
 
         // Spin locking on the same lock always works, assuming no concurrent access.
-        let acquired = lock.spin_lock(None).await.unwrap();
+        let guard = lock.spin_lock(None).await.unwrap();
 
         // Releasing still works.
-        release_lock(Some(acquired)).await;
+        release_lock(CrossProcessLockResult::Clean(guard)).await;
         assert_eq!(lock.num_holders.load(atomic::Ordering::SeqCst), 0);
 
         Ok(())
@@ -585,8 +600,8 @@ mod tests {
         let lock = CrossProcessLock::new(store.clone(), "key".to_owned(), "first".to_owned());
 
         // When a lock is acquired...
-        let acquired = lock.try_lock_once().await?;
-        assert!(acquired.is_some());
+        let result = lock.try_lock_once().await?;
+        assert!(result.is_ok());
         assert_eq!(lock.num_holders.load(atomic::Ordering::SeqCst), 1);
 
         // But then forgotten... (note: no need to release the guard)
@@ -596,8 +611,8 @@ mod tests {
         let lock = CrossProcessLock::new(store.clone(), "key".to_owned(), "first".to_owned());
 
         // We still got it.
-        let acquired = lock.try_lock_once().await?;
-        assert!(acquired.is_some());
+        let result = lock.try_lock_once().await?;
+        assert!(result.is_ok());
         assert_eq!(lock.num_holders.load(atomic::Ordering::SeqCst), 1);
 
         Ok(())
@@ -609,19 +624,19 @@ mod tests {
         let lock = CrossProcessLock::new(store, "key".to_owned(), "first".to_owned());
 
         // Taking the lock twice...
-        let acquired = lock.try_lock_once().await?;
-        assert!(acquired.is_some());
+        let result1 = lock.try_lock_once().await?;
+        assert!(result1.is_ok());
 
-        let acquired2 = lock.try_lock_once().await?;
-        assert!(acquired2.is_some());
+        let result2 = lock.try_lock_once().await?;
+        assert!(result2.is_ok());
 
         assert_eq!(lock.num_holders.load(atomic::Ordering::SeqCst), 2);
 
         // ...means we can release it twice.
-        release_lock(acquired).await;
+        release_lock(result1).await;
         assert_eq!(lock.num_holders.load(atomic::Ordering::SeqCst), 1);
 
-        release_lock(acquired2).await;
+        release_lock(result2).await;
         assert_eq!(lock.num_holders.load(atomic::Ordering::SeqCst), 0);
 
         Ok(())
@@ -634,22 +649,22 @@ mod tests {
         let lock2 = CrossProcessLock::new(store, "key".to_owned(), "second".to_owned());
 
         // When the first process takes the lock...
-        let acquired1 = lock1.try_lock_once().await?;
-        assert!(acquired1.is_some());
+        let result1 = lock1.try_lock_once().await?;
+        assert!(result1.is_ok());
 
         // The second can't take it immediately.
-        let acquired2 = lock2.try_lock_once().await?;
-        assert!(acquired2.is_none());
+        let result2 = lock2.try_lock_once().await?;
+        assert!(result2.is_ok().not());
 
         let lock2_clone = lock2.clone();
         let handle = spawn(async move { lock2_clone.spin_lock(Some(1000)).await });
 
         sleep(Duration::from_millis(100)).await;
 
-        drop(acquired1);
+        drop(result1);
 
         // lock2 in the background manages to get the lock at some point.
-        let _acquired2 = handle
+        let _result2 = handle
             .await
             .expect("join handle is properly awaited")
             .expect("lock was obtained after spin-locking");
