feat: Event cache lock has a _generation_ value.

This patch introduces the _generation_ value for the event cache lock:
every time the lock is acquired by a new holder, the generation is
incremented by 1. This is fundamental to know if the lock has been
poisoned or not.

Author: Hywan

crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs

@@ -260,57 +260,57 @@ macro_rules! event_cache_store_integration_tests_time {
                 let store = get_event_cache_store().await.unwrap().into_event_cache_store();
 
                 let acquired0 = store.try_take_leased_lock(0, "key", "alice").await.unwrap();
-                assert!(acquired0);
+                assert_eq!(acquired0, Some(0)); // first lock generation
 
                 // Should extend the lease automatically (same holder).
                 let acquired2 = store.try_take_leased_lock(300, "key", "alice").await.unwrap();
-                assert!(acquired2);
+                assert_eq!(acquired2, Some(0)); // same lock generation
 
                 // Should extend the lease automatically (same holder + time is ok).
                 let acquired3 = store.try_take_leased_lock(300, "key", "alice").await.unwrap();
-                assert!(acquired3);
+                assert_eq!(acquired3, Some(0)); // same lock generation
 
                 // Another attempt at taking the lock should fail, because it's taken.
                 let acquired4 = store.try_take_leased_lock(300, "key", "bob").await.unwrap();
-                assert!(!acquired4);
+                assert!(acquired4.is_none()); // not acquired
 
                 // Even if we insist.
                 let acquired5 = store.try_take_leased_lock(300, "key", "bob").await.unwrap();
-                assert!(!acquired5);
+                assert!(acquired5.is_none()); // not acquired
 
                 // That's a nice test we got here, go take a little nap.
                 tokio::time::sleep(Duration::from_millis(50)).await;
 
                 // Still too early.
                 let acquired55 = store.try_take_leased_lock(300, "key", "bob").await.unwrap();
-                assert!(!acquired55);
+                assert!(acquired55.is_none()); // not acquired
 
                 // Ok you can take another nap then.
                 tokio::time::sleep(Duration::from_millis(250)).await;
 
                 // At some point, we do get the lock.
                 let acquired6 = store.try_take_leased_lock(0, "key", "bob").await.unwrap();
-                assert!(acquired6);
+                assert_eq!(acquired6, Some(1)); // new lock generation!
 
                 tokio::time::sleep(Duration::from_millis(1)).await;
 
                 // The other gets it almost immediately too.
                 let acquired7 = store.try_take_leased_lock(0, "key", "alice").await.unwrap();
-                assert!(acquired7);
+                assert_eq!(acquired7, Some(2)); // new lock generation!
 
                 tokio::time::sleep(Duration::from_millis(1)).await;
 
-                // But when we take a longer lease...
+                // But when we take a longer lease…
                 let acquired8 = store.try_take_leased_lock(300, "key", "bob").await.unwrap();
-                assert!(acquired8);
+                assert_eq!(acquired8, Some(3)); // new lock generation!
 
                 // It blocks the other user.
                 let acquired9 = store.try_take_leased_lock(300, "key", "alice").await.unwrap();
-                assert!(!acquired9);
+                assert!(acquired9.is_none()); // not acquired
 
                 // We can hold onto our lease.
                 let acquired10 = store.try_take_leased_lock(300, "key", "bob").await.unwrap();
-                assert!(acquired10);
+                assert_eq!(acquired10, Some(3)); // same lock generation
             }
         }
     };

crates/matrix-sdk-base/src/event_cache/store/memory_store.rs

@@ -16,7 +16,8 @@ use std::{collections::HashMap, num::NonZeroUsize, sync::RwLock as StdRwLock, ti
 
 use async_trait::async_trait;
 use matrix_sdk_common::{
-    ring_buffer::RingBuffer, store_locks::memory_store_helper::try_take_leased_lock,
+    ring_buffer::RingBuffer,
+    store_locks::{memory_store_helper::try_take_leased_lock, LockGeneration},
 };
 use ruma::{MxcUri, OwnedMxcUri};
 
@@ -30,7 +31,7 @@ use crate::media::{MediaRequestParameters, UniqueKey as _};
 #[derive(Debug)]
 pub struct MemoryStore {
     media: StdRwLock<RingBuffer<(OwnedMxcUri, String /* unique key */, Vec<u8>)>>,
-    leases: StdRwLock<HashMap<String, (String, Instant)>>,
+    leases: StdRwLock<HashMap<String, (String, Instant, LockGeneration)>>,
 }
 
 // SAFETY: `new_unchecked` is safe because 20 is not zero.
@@ -62,7 +63,7 @@ impl EventCacheStore for MemoryStore {
         lease_duration_ms: u32,
         key: &str,
         holder: &str,
-    ) -> Result<bool, Self::Error> {
+    ) -> Result<Option<LockGeneration>, Self::Error> {
         Ok(try_take_leased_lock(&self.leases, lease_duration_ms, key, holder))
     }
 

crates/matrix-sdk-base/src/event_cache/store/mod.rs

@@ -19,7 +19,7 @@
 //! into the event cache for the actual storage. By default this brings an
 //! in-memory store.
 
-use std::{fmt, ops::Deref, str::Utf8Error, sync::Arc};
+use std::{fmt, ops::Deref, result::Result as StdResult, str::Utf8Error, sync::Arc};
 
 #[cfg(any(test, feature = "testing"))]
 #[macro_use]
@@ -160,7 +160,7 @@ impl EventCacheStoreError {
 }
 
 /// An `EventCacheStore` specific result type.
-pub type Result<T, E = EventCacheStoreError> = std::result::Result<T, E>;
+pub type Result<T, E = EventCacheStoreError> = StdResult<T, E>;
 
 /// A type that wraps the [`EventCacheStore`] but implements [`BackingStore`] to
 /// make it usable inside the cross process lock.
@@ -177,7 +177,7 @@ impl BackingStore for LockableEventCacheStore {
         lease_duration_ms: u32,
         key: &str,
         holder: &str,
-    ) -> std::result::Result<bool, Self::LockError> {
-        self.0.try_take_leased_lock(lease_duration_ms, key, holder).await
+    ) -> StdResult<bool, Self::LockError> {
+        Ok(self.0.try_take_leased_lock(lease_duration_ms, key, holder).await?.is_some())
     }
 }

crates/matrix-sdk-base/src/event_cache/store/traits.rs

@@ -15,7 +15,7 @@
 use std::{fmt, sync::Arc};
 
 use async_trait::async_trait;
-use matrix_sdk_common::AsyncTraitDeps;
+use matrix_sdk_common::{store_locks::LockGeneration, AsyncTraitDeps};
 use ruma::MxcUri;
 
 use super::EventCacheStoreError;
@@ -35,7 +35,7 @@ pub trait EventCacheStore: AsyncTraitDeps {
         lease_duration_ms: u32,
         key: &str,
         holder: &str,
-    ) -> Result<bool, Self::Error>;
+    ) -> Result<Option<LockGeneration>, Self::Error>;
 
     /// Add a media file's content in the media store.
     ///
@@ -127,7 +127,7 @@ impl<T: EventCacheStore> EventCacheStore for EraseEventCacheStoreError<T> {
         lease_duration_ms: u32,
         key: &str,
         holder: &str,
-    ) -> Result<bool, Self::Error> {
+    ) -> Result<Option<LockGeneration>, Self::Error> {
         self.0.try_take_leased_lock(lease_duration_ms, key, holder).await.map_err(Into::into)
     }
 

crates/matrix-sdk-common/src/store_locks.rs

@@ -54,6 +54,13 @@ use crate::{
     SendOutsideWasm,
 };
 
+/// A lock generation is an integer incremented each time it is taken by another
+/// holder. This is not used by all cross-process locks.
+pub type LockGeneration = u64;
+
+/// Describe the first lock generation value (see [`LockGeneration`]).
+pub const FIRST_LOCK_GENERATION: u64 = 0;
+
 /// Backing store for a cross-process lock.
 #[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
@@ -351,16 +358,23 @@ mod tests {
 
     use super::{
         memory_store_helper::try_take_leased_lock, BackingStore, CrossProcessStoreLock,
-        CrossProcessStoreLockGuard, LockStoreError, EXTEND_LEASE_EVERY_MS,
+        CrossProcessStoreLockGuard, LockGeneration, LockStoreError, EXTEND_LEASE_EVERY_MS,
     };
 
+    type HolderExpirationGeneration = (String, Instant, LockGeneration);
+
     #[derive(Clone, Default)]
     struct TestStore {
-        leases: Arc<RwLock<HashMap<String, (String, Instant)>>>,
+        leases: Arc<RwLock<HashMap<String, HolderExpirationGeneration>>>,
     }
 
     impl TestStore {
-        fn try_take_leased_lock(&self, lease_duration_ms: u32, key: &str, holder: &str) -> bool {
+        fn try_take_leased_lock(
+            &self,
+            lease_duration_ms: u32,
+            key: &str,
+            holder: &str,
+        ) -> Option<LockGeneration> {
             try_take_leased_lock(&self.leases, lease_duration_ms, key, holder)
         }
     }
@@ -380,7 +394,7 @@ mod tests {
             key: &str,
             holder: &str,
         ) -> Result<bool, Self::LockError> {
-            Ok(self.try_take_leased_lock(lease_duration_ms, key, holder))
+            Ok(self.try_take_leased_lock(lease_duration_ms, key, holder).is_some())
         }
     }
 
@@ -506,36 +520,45 @@ pub mod memory_store_helper {
         time::{Duration, Instant},
     };
 
+    use super::LockGeneration;
+
+    type HolderExpirationGeneration = (String, Instant, LockGeneration);
+
+    /// Try to acquire or to extend the lock.
+    ///
+    /// Return `Some` if the lock has been acquired (or extended). It contains
+    /// the generation number.
     pub fn try_take_leased_lock(
-        leases: &RwLock<HashMap<String, (String, Instant)>>,
+        leases: &RwLock<HashMap<String, HolderExpirationGeneration>>,
         lease_duration_ms: u32,
         key: &str,
         holder: &str,
-    ) -> bool {
+    ) -> Option<LockGeneration> {
         let now = Instant::now();
         let expiration = now + Duration::from_millis(lease_duration_ms.into());
 
         match leases.write().unwrap().entry(key.to_owned()) {
             // There is an existing holder.
             Entry::Occupied(mut entry) => {
-                let (current_holder, current_expiration) = entry.get_mut();
+                let (current_holder, current_expiration, current_generation) = entry.get_mut();
 
                 if current_holder == holder {
                     // We had the lease before, extend it.
                     *current_expiration = expiration;
 
-                    true
+                    Some(*current_generation)
                 } else {
                     // We didn't have it.
                     if *current_expiration < now {
                         // Steal it!
                         *current_holder = holder.to_owned();
                         *current_expiration = expiration;
+                        *current_generation += 1;
 
-                        true
+                        Some(*current_generation)
                     } else {
                         // We tried our best.
-                        false
+                        None
                     }
                 }
             }
@@ -545,9 +568,10 @@ pub mod memory_store_helper {
                 entry.insert((
                     holder.to_owned(),
                     Instant::now() + Duration::from_millis(lease_duration_ms.into()),
+                    0,
                 ));
 
-                true
+                Some(0)
             }
         }
     }

crates/matrix-sdk-crypto/src/store/memorystore.rs

@@ -20,7 +20,7 @@ use std::{
 };
 
 use async_trait::async_trait;
-use matrix_sdk_common::store_locks::memory_store_helper::try_take_leased_lock;
+use matrix_sdk_common::store_locks::{memory_store_helper::try_take_leased_lock, LockGeneration};
 use ruma::{
     events::secret::request::SecretName, DeviceId, OwnedDeviceId, OwnedRoomId, OwnedTransactionId,
     OwnedUserId, RoomId, TransactionId, UserId,
@@ -90,7 +90,7 @@ pub struct MemoryStore {
     key_requests_by_info: StdRwLock<HashMap<String, OwnedTransactionId>>,
     direct_withheld_info: StdRwLock<HashMap<OwnedRoomId, HashMap<String, RoomKeyWithheldEvent>>>,
     custom_values: StdRwLock<HashMap<String, Vec<u8>>>,
-    leases: StdRwLock<HashMap<String, (String, Instant)>>,
+    leases: StdRwLock<HashMap<String, (String, Instant, LockGeneration)>>,
     secret_inbox: StdRwLock<HashMap<String, Vec<GossippedSecret>>>,
     backup_keys: RwLock<BackupKeys>,
     next_batch_token: RwLock<Option<String>>,
@@ -632,7 +632,7 @@ impl CryptoStore for MemoryStore {
         key: &str,
         holder: &str,
     ) -> Result<bool> {
-        Ok(try_take_leased_lock(&self.leases, lease_duration_ms, key, holder))
+        Ok(try_take_leased_lock(&self.leases, lease_duration_ms, key, holder).is_some())
     }
 }
 

crates/matrix-sdk-sqlite/migrations/event_cache_store/003_lease_locks_with_generation.sql

@@ -0,0 +1 @@
+ALTER TABLE "lease_locks" ADD COLUMN "generation" INTEGER NOT NULL DEFAULT 0;

crates/matrix-sdk-sqlite/src/event_cache_store.rs

@@ -5,6 +5,7 @@ use deadpool_sqlite::{Object as SqliteAsyncConn, Pool as SqlitePool, Runtime};
 use matrix_sdk_base::{
     event_cache::store::EventCacheStore,
     media::{MediaRequestParameters, UniqueKey},
+    store_locks::LockGeneration,
 };
 use matrix_sdk_store_encryption::StoreCipher;
 use ruma::MilliSecondsSinceUnixEpoch;
@@ -28,7 +29,7 @@ mod keys {
 /// This is used to figure whether the SQLite database requires a migration.
 /// Every new SQL migration should imply a bump of this number, and changes in
 /// the [`run_migrations`] function.
-const DATABASE_VERSION: u8 = 2;
+const DATABASE_VERSION: u8 = 3;
 
 /// A SQLite-based event cache store.
 #[derive(Clone)]
@@ -142,6 +143,16 @@ async fn run_migrations(conn: &SqliteAsyncConn, version: u8) -> Result<()> {
         .await?;
     }
 
+    if version < 3 {
+        conn.with_transaction(|txn| {
+            txn.execute_batch(include_str!(
+                "../migrations/event_cache_store/003_lease_locks_with_generation.sql"
+            ))?;
+            txn.set_db_version(3)
+        })
+        .await?;
+    }
+
     Ok(())
 }
 
@@ -154,32 +165,41 @@ impl EventCacheStore for SqliteEventCacheStore {
         lease_duration_ms: u32,
         key: &str,
         holder: &str,
-    ) -> Result<bool> {
+    ) -> Result<Option<LockGeneration>> {
         let key = key.to_owned();
         let holder = holder.to_owned();
 
         let now: u64 = MilliSecondsSinceUnixEpoch::now().get().into();
         let expiration = now + lease_duration_ms as u64;
 
-        let num_touched = self
+        let generation = self
             .acquire()
             .await?
             .with_transaction(move |txn| {
-                txn.execute(
+                txn.query_row(
                     "INSERT INTO lease_locks (key, holder, expiration)
                     VALUES (?1, ?2, ?3)
-                    ON CONFLICT (key)
-                    DO
-                        UPDATE SET holder = ?2, expiration = ?3
-                        WHERE holder = ?2
+                    ON CONFLICT (key) DO
+                    UPDATE SET
+                        holder = excluded.holder,
+                        expiration = excluded.expiration,
+                        generation =
+                            CASE holder
+                                WHEN excluded.holder THEN generation
+                                ELSE generation + 1
+                            END
+                    WHERE
+                        holder = excluded.holder
                         OR expiration < ?4
-                ",
+                    RETURNING generation",
                     (key, holder, expiration, now),
+                    |row| -> rusqlite::Result<u64> { row.get(0) },
                 )
+                .optional()
             })
             .await?;
 
-        Ok(num_touched == 1)
+        Ok(generation)
     }
 
     async fn add_media_content(

crates/matrix-sdk/src/event_cache/mod.rs

@@ -225,6 +225,7 @@ impl EventCache {
         mut room_updates_feed: Receiver<RoomUpdates>,
     ) {
         trace!("Spawning the listen task");
+
         loop {
             match room_updates_feed.recv().await {
                 Ok(updates) => {