From 62801f1a6ca72aea2baa77aaad0a221d557f22e1 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Tue, 25 Jun 2024 13:55:29 +0200 Subject: [PATCH] state store: change schema for send_queue_events table It turns out that so as to be able to read the room ids, they need to be *values*, not only *keys* (since keys are one-way hashed). This means we need to duplicate the room_id field in indexeddb/sqlite, so each entry contains both the room_id as a key (for queries) and as a value (to return it). Since there's no meaningful migration we can apply, the way to go is to drop the pending events table and recreate it from the ground up. It is assumed that no one has used the store on indexeddb; otherwise the workaround would be to drop and recreate it. --- .../src/store/integration_tests.rs | 32 ++++++++++ .../matrix-sdk-base/src/store/memory_store.rs | 24 ++++++-- crates/matrix-sdk-base/src/store/traits.rs | 12 +++- .../src/state_store/mod.rs | 60 ++++++++++++++++--- .../004_send_queue_with_roomid_value.sql | 22 +++++++ crates/matrix-sdk-sqlite/src/state_store.rs | 45 ++++++++++++-- 6 files changed, 174 insertions(+), 21 deletions(-) create mode 100644 crates/matrix-sdk-sqlite/migrations/state_store/004_send_queue_with_roomid_value.sql diff --git a/crates/matrix-sdk-base/src/store/integration_tests.rs b/crates/matrix-sdk-base/src/store/integration_tests.rs index 569c9c1d7..23a70e635 100644 --- a/crates/matrix-sdk-base/src/store/integration_tests.rs +++ b/crates/matrix-sdk-base/src/store/integration_tests.rs @@ -1385,6 +1385,38 @@ impl StateStoreIntegrationTests for DynStateStore { for i in 0..3 { assert_ne!(pending[i].transaction_id, txn0); } + + // Now add one event for two other rooms, remove one of the events, and then + // query all the rooms which have outstanding unsent events. + + // Add one event for room2. + let room_id2 = room_id!("!test_send_queue_two:localhost"); + { + let txn = TransactionId::new(); + let event = + SerializableEventContent::new(RoomMessageEventContent::text_plain("room2").into()) + .unwrap(); + self.save_send_queue_event(room_id2, txn.clone(), event).await.unwrap(); + } + + // Add and remove one event for room3. + { + let room_id3 = room_id!("!test_send_queue_three:localhost"); + let txn = TransactionId::new(); + let event = + SerializableEventContent::new(RoomMessageEventContent::text_plain("room3").into()) + .unwrap(); + self.save_send_queue_event(room_id3, txn.clone(), event).await.unwrap(); + + self.remove_send_queue_event(room_id3, &txn).await.unwrap(); + } + + // Query all the rooms which have unsent events. Per the previous steps, + // it should be room1 and room2, not room3. + let outstanding_rooms = self.load_rooms_with_unsent_events().await.unwrap(); + assert_eq!(outstanding_rooms.len(), 2); + assert!(outstanding_rooms.iter().any(|room| room == room_id)); + assert!(outstanding_rooms.iter().any(|room| room == room_id2)); } } diff --git a/crates/matrix-sdk-base/src/store/memory_store.rs b/crates/matrix-sdk-base/src/store/memory_store.rs index b27220e43..9c5d92188 100644 --- a/crates/matrix-sdk-base/src/store/memory_store.rs +++ b/crates/matrix-sdk-base/src/store/memory_store.rs @@ -890,12 +890,20 @@ impl StateStore for MemoryStore { room_id: &RoomId, transaction_id: &TransactionId, ) -> Result<(), Self::Error> { - self.send_queue_events - .write() - .unwrap() - .entry(room_id.to_owned()) - .or_default() - .retain(|item| item.transaction_id != transaction_id); + let mut q = self.send_queue_events.write().unwrap(); + + let entry = q.get_mut(room_id); + if let Some(entry) = entry { + // Find the event by id in its room queue, and remove it if present. + if let Some(pos) = entry.iter().position(|item| item.transaction_id == transaction_id) { + entry.remove(pos); + // And if this was the last event before removal, remove the entire room entry. + if entry.is_empty() { + q.remove(room_id); + } + } + } + Ok(()) } @@ -925,6 +933,10 @@ impl StateStore for MemoryStore { } Ok(()) } + + async fn load_rooms_with_unsent_events(&self) -> Result, Self::Error> { + Ok(self.send_queue_events.read().unwrap().keys().cloned().collect()) + } } #[cfg(test)] diff --git a/crates/matrix-sdk-base/src/store/traits.rs b/crates/matrix-sdk-base/src/store/traits.rs index b8214eeb2..c640a50db 100644 --- a/crates/matrix-sdk-base/src/store/traits.rs +++ b/crates/matrix-sdk-base/src/store/traits.rs @@ -34,7 +34,8 @@ use ruma::{ StateEventType, StaticEventContent, StaticStateEventContent, }, serde::Raw, - EventId, MxcUri, OwnedEventId, OwnedTransactionId, OwnedUserId, RoomId, TransactionId, UserId, + EventId, MxcUri, OwnedEventId, OwnedRoomId, OwnedTransactionId, OwnedUserId, RoomId, + TransactionId, UserId, }; use serde::{Deserialize, Serialize}; @@ -429,6 +430,9 @@ pub trait StateStore: AsyncTraitDeps { transaction_id: &TransactionId, wedged: bool, ) -> Result<(), Self::Error>; + + /// Loads all the rooms which have any pending events in their send queue. + async fn load_rooms_with_unsent_events(&self) -> Result, Self::Error>; } #[repr(transparent)] @@ -688,6 +692,10 @@ impl StateStore for EraseStateStoreError { .await .map_err(Into::into) } + + async fn load_rooms_with_unsent_events(&self) -> Result, Self::Error> { + self.0.load_rooms_with_unsent_events().await.map_err(Into::into) + } } /// Convenience functionality for state stores. @@ -1053,7 +1061,7 @@ impl SerializableEventContent { } /// An event to be sent with a send queue. -#[derive(Clone, Serialize, Deserialize)] +#[derive(Clone)] pub struct QueuedEvent { /// The content of the message-like event we'd like to send. pub event: SerializableEventContent, diff --git a/crates/matrix-sdk-indexeddb/src/state_store/mod.rs b/crates/matrix-sdk-indexeddb/src/state_store/mod.rs index 71abe53f8..a8a469925 100644 --- a/crates/matrix-sdk-indexeddb/src/state_store/mod.rs +++ b/crates/matrix-sdk-indexeddb/src/state_store/mod.rs @@ -44,8 +44,8 @@ use ruma::{ GlobalAccountDataEventType, RoomAccountDataEventType, StateEventType, SyncStateEvent, }, serde::Raw, - CanonicalJsonObject, EventId, MxcUri, OwnedEventId, OwnedTransactionId, OwnedUserId, RoomId, - RoomVersionId, TransactionId, UserId, + CanonicalJsonObject, EventId, MxcUri, OwnedEventId, OwnedRoomId, OwnedTransactionId, + OwnedUserId, RoomId, RoomVersionId, TransactionId, UserId, }; use serde::{de::DeserializeOwned, Deserialize, Serialize}; use tracing::{debug, warn}; @@ -408,6 +408,19 @@ impl IndexeddbStateStore { } } +/// A superset of [`QueuedEvent`] that also contains the room id, since we want +/// to return them. +#[derive(Serialize, Deserialize)] +struct PersistedQueuedEvent { + /// In which room is this event going to be sent. + pub room_id: OwnedRoomId, + + // All these fields are the same as in [`QueuedEvent`]. + event: SerializableEventContent, + transaction_id: OwnedTransactionId, + is_wedged: bool, +} + // Small hack to have the following macro invocation act as the appropriate // trait impl block on wasm, but still be compiled on non-wasm as a regular // impl block otherwise. @@ -1363,11 +1376,16 @@ impl_state_store!({ let mut prev = prev.map_or_else( || Ok(Vec::new()), - |val| self.deserialize_value::>(&val), + |val| self.deserialize_value::>(&val), )?; // Push the new event. - prev.push(QueuedEvent { event: content, transaction_id, is_wedged: false }); + prev.push(PersistedQueuedEvent { + room_id: room_id.to_owned(), + event: content, + transaction_id, + is_wedged: false, + }); // Save the new vector into db. obj.put_key_val(&encoded_key, &self.serialize_value(&prev)?)?; @@ -1394,7 +1412,7 @@ impl_state_store!({ // Reload the previous vector for this room. if let Some(val) = obj.get(&encoded_key)?.await? { - let mut prev = self.deserialize_value::>(&val)?; + let mut prev = self.deserialize_value::>(&val)?; if let Some(pos) = prev.iter().position(|item| item.transaction_id == transaction_id) { prev.remove(pos); @@ -1424,10 +1442,17 @@ impl_state_store!({ let prev = prev.map_or_else( || Ok(Vec::new()), - |val| self.deserialize_value::>(&val), + |val| self.deserialize_value::>(&val), )?; - Ok(prev) + Ok(prev + .into_iter() + .map(|item| QueuedEvent { + event: item.event, + transaction_id: item.transaction_id, + is_wedged: item.is_wedged, + }) + .collect()) } async fn update_send_queue_event_status( @@ -1445,7 +1470,7 @@ impl_state_store!({ let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?; if let Some(val) = obj.get(&encoded_key)?.await? { - let mut prev = self.deserialize_value::>(&val)?; + let mut prev = self.deserialize_value::>(&val)?; if let Some(queued_event) = prev.iter_mut().find(|item| item.transaction_id == transaction_id) { @@ -1458,6 +1483,25 @@ impl_state_store!({ Ok(()) } + + async fn load_rooms_with_unsent_events(&self) -> Result> { + let tx = self + .inner + .transaction_on_one_with_mode(keys::ROOM_SEND_QUEUE, IdbTransactionMode::Readwrite)?; + + let obj = tx.object_store(keys::ROOM_SEND_QUEUE)?; + + let all_entries = obj + .get_all()? + .await? + .into_iter() + .map(|item| { + self.deserialize_value(&item).map(|event: PersistedQueuedEvent| event.room_id) + }) + .collect::, _>>()?; + + Ok(all_entries.into_iter().collect()) + } }); /// A room member. diff --git a/crates/matrix-sdk-sqlite/migrations/state_store/004_send_queue_with_roomid_value.sql b/crates/matrix-sdk-sqlite/migrations/state_store/004_send_queue_with_roomid_value.sql new file mode 100644 index 000000000..6516a5601 --- /dev/null +++ b/crates/matrix-sdk-sqlite/migrations/state_store/004_send_queue_with_roomid_value.sql @@ -0,0 +1,22 @@ +-- Send queue events, keyed by room id and transaction id, but also include the room id in the +-- value. +DROP TABLE "send_queue_events"; + +CREATE TABLE "send_queue_events" ( + -- This is used as a key, thus hashed. + "room_id" BLOB NOT NULL, + + -- This is used as a value (thus encrypted/decrypted). + "room_id_val" BLOB NOT NULL, + + -- This is used as both a key and a value, thus neither encrypted/decrypted/hashed. + "transaction_id" BLOB NOT NULL, + + -- Used as a value, thus encrypted/decrypted. + "content" BLOB NOT NULL, + + -- In clear. + "wedged" BOOLEAN NOT NULL, + + PRIMARY KEY ("room_id", "transaction_id") +); diff --git a/crates/matrix-sdk-sqlite/src/state_store.rs b/crates/matrix-sdk-sqlite/src/state_store.rs index 5fc8cea5a..904633217 100644 --- a/crates/matrix-sdk-sqlite/src/state_store.rs +++ b/crates/matrix-sdk-sqlite/src/state_store.rs @@ -29,8 +29,8 @@ use ruma::{ GlobalAccountDataEventType, RoomAccountDataEventType, StateEventType, }, serde::Raw, - CanonicalJsonObject, EventId, OwnedEventId, OwnedTransactionId, OwnedUserId, RoomId, - RoomVersionId, TransactionId, UserId, + CanonicalJsonObject, EventId, OwnedEventId, OwnedRoomId, OwnedTransactionId, OwnedUserId, + RoomId, RoomVersionId, TransactionId, UserId, }; use rusqlite::{OptionalExtension, Transaction}; use serde::{de::DeserializeOwned, Deserialize, Serialize}; @@ -59,7 +59,7 @@ mod keys { pub const SEND_QUEUE: &str = "send_queue_events"; } -const DATABASE_VERSION: u8 = 4; +const DATABASE_VERSION: u8 = 5; /// A sqlite based cryptostore. #[derive(Clone)] @@ -224,6 +224,17 @@ impl SqliteStateStore { .await?; } + if from < 5 && to >= 5 { + conn.with_transaction(move |txn| { + // Create new table. + txn.execute_batch(include_str!( + "../migrations/state_store/004_send_queue_with_roomid_value.sql" + ))?; + Result::<_, Error>::Ok(()) + }) + .await?; + } + conn.set_kv("version", vec![to]).await?; Ok(()) @@ -1674,7 +1685,8 @@ impl StateStore for SqliteStateStore { transaction_id: OwnedTransactionId, content: SerializableEventContent, ) -> Result<(), Self::Error> { - let room_id = self.encode_key(keys::SEND_QUEUE, room_id); + let room_id_key = self.encode_key(keys::SEND_QUEUE, room_id); + let room_id_value = self.serialize_value(&room_id.to_owned())?; let content = self.serialize_json(&content)?; @@ -1686,7 +1698,7 @@ impl StateStore for SqliteStateStore { self.acquire() .await? .with_transaction(move |txn| { - txn.prepare_cached("INSERT INTO send_queue_events (room_id, transaction_id, content, wedged) VALUES (?, ?, ?, false)")?.execute((room_id, transaction_id.to_string(), content))?; + txn.prepare_cached("INSERT INTO send_queue_events (room_id, room_id_val, transaction_id, content, wedged) VALUES (?, ?, ?, ?, false)")?.execute((room_id_key, room_id_value, transaction_id.to_string(), content))?; Ok(()) }) .await @@ -1767,6 +1779,29 @@ impl StateStore for SqliteStateStore { }) .await } + + async fn load_rooms_with_unsent_events(&self) -> Result, Self::Error> { + // If the values were not encrypted, we could use `SELECT DISTINCT` here, but we + // have to manually do the deduplication: indeed, for all X, encrypt(X) + // != encrypted(X), since we use a nonce in the encryption process. + + let res: Vec> = self + .acquire() + .await? + .prepare("SELECT room_id_val FROM send_queue_events", |mut stmt| { + stmt.query(())?.mapped(|row| Ok(row.get(0)?)).collect() + }) + .await?; + + // So we collect the results into a `BTreeSet` to perform the deduplication, and + // then rejigger that into a vector. + Ok(res + .into_iter() + .map(|entry| self.deserialize_value(&entry)) + .collect::, _>>()? + .into_iter() + .collect()) + } } #[derive(Debug, Clone, Serialize, Deserialize)]