diff --git a/crates/matrix-sdk-base/src/store/integration_tests.rs b/crates/matrix-sdk-base/src/store/integration_tests.rs index 533a6f064..dc1ae7b69 100644 --- a/crates/matrix-sdk-base/src/store/integration_tests.rs +++ b/crates/matrix-sdk-base/src/store/integration_tests.rs @@ -1498,14 +1498,21 @@ impl StateStoreIntegrationTests for DynStateStore { assert!(self.list_dependent_send_queue_events(room_id).await.unwrap().is_empty()); // Save a redaction for that event. - self.save_dependent_send_queue_event(room_id, &txn0, DependentQueuedEventKind::Redact) - .await - .unwrap(); + let child_txn = TransactionId::new(); + self.save_dependent_send_queue_event( + room_id, + &txn0, + child_txn.clone(), + DependentQueuedEventKind::Redact, + ) + .await + .unwrap(); // It worked. let dependents = self.list_dependent_send_queue_events(room_id).await.unwrap(); assert_eq!(dependents.len(), 1); - assert_eq!(dependents[0].transaction_id, txn0); + assert_eq!(dependents[0].parent_transaction_id, txn0); + assert_eq!(dependents[0].own_transaction_id, child_txn); assert!(dependents[0].event_id.is_none()); assert_matches!(dependents[0].kind, DependentQueuedEventKind::Redact); @@ -1518,13 +1525,16 @@ impl StateStoreIntegrationTests for DynStateStore { // It worked. let dependents = self.list_dependent_send_queue_events(room_id).await.unwrap(); assert_eq!(dependents.len(), 1); - assert_eq!(dependents[0].transaction_id, txn0); + assert_eq!(dependents[0].parent_transaction_id, txn0); + assert_eq!(dependents[0].own_transaction_id, child_txn); assert_eq!(dependents[0].event_id.as_ref(), Some(&event_id)); assert_matches!(dependents[0].kind, DependentQueuedEventKind::Redact); // Now remove it. - let removed = - self.remove_dependent_send_queue_event(room_id, dependents[0].id).await.unwrap(); + let removed = self + .remove_dependent_send_queue_event(room_id, &dependents[0].own_transaction_id) + .await + .unwrap(); assert!(removed); // It worked. @@ -1538,14 +1548,20 @@ impl StateStoreIntegrationTests for DynStateStore { .unwrap(); self.save_send_queue_event(room_id, txn1.clone(), event1).await.unwrap(); - self.save_dependent_send_queue_event(room_id, &txn0, DependentQueuedEventKind::Redact) - .await - .unwrap(); + self.save_dependent_send_queue_event( + room_id, + &txn0, + TransactionId::new(), + DependentQueuedEventKind::Redact, + ) + .await + .unwrap(); assert_eq!(self.list_dependent_send_queue_events(room_id).await.unwrap().len(), 1); self.save_dependent_send_queue_event( room_id, &txn1, + TransactionId::new(), DependentQueuedEventKind::Edit { new_content: SerializableEventContent::new( &RoomMessageEventContent::text_plain("edit").into(), diff --git a/crates/matrix-sdk-base/src/store/memory_store.rs b/crates/matrix-sdk-base/src/store/memory_store.rs index 15fe6f5ce..bfe44afb2 100644 --- a/crates/matrix-sdk-base/src/store/memory_store.rs +++ b/crates/matrix-sdk-base/src/store/memory_store.rs @@ -15,7 +15,7 @@ use std::{ collections::{BTreeMap, BTreeSet, HashMap}, num::NonZeroUsize, - sync::{Mutex, RwLock as StdRwLock}, + sync::RwLock as StdRwLock, }; use async_trait::async_trait; @@ -92,7 +92,6 @@ pub struct MemoryStore { custom: StdRwLock, Vec>>, send_queue_events: StdRwLock>>, dependent_send_queue_events: StdRwLock>>, - dependent_send_queue_event_next_id: Mutex, } // SAFETY: `new_unchecked` is safe because 20 is not zero. @@ -124,7 +123,6 @@ impl Default for MemoryStore { custom: Default::default(), send_queue_events: Default::default(), dependent_send_queue_events: Default::default(), - dependent_send_queue_event_next_id: Mutex::new(0), } } } @@ -1005,39 +1003,31 @@ impl StateStore for MemoryStore { async fn save_dependent_send_queue_event( &self, room: &RoomId, - transaction_id: &TransactionId, + parent_transaction_id: &TransactionId, + own_transaction_id: OwnedTransactionId, content: DependentQueuedEventKind, ) -> Result<(), Self::Error> { - let id = { - let mut next_id = self.dependent_send_queue_event_next_id.lock().unwrap(); - // Don't tell anyone, but sometimes I miss C++'s `x++` operator. - let id = *next_id; - *next_id += 1; - id - }; - self.dependent_send_queue_events.write().unwrap().entry(room.to_owned()).or_default().push( DependentQueuedEvent { - id, kind: content, - transaction_id: transaction_id.to_owned(), + parent_transaction_id: parent_transaction_id.to_owned(), + own_transaction_id, event_id: None, }, ); - Ok(()) } async fn update_dependent_send_queue_event( &self, room: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, event_id: OwnedEventId, ) -> Result { let mut dependent_send_queue_events = self.dependent_send_queue_events.write().unwrap(); let dependents = dependent_send_queue_events.entry(room.to_owned()).or_default(); let mut num_updated = 0; - for d in dependents.iter_mut().filter(|item| item.transaction_id == transaction_id) { + for d in dependents.iter_mut().filter(|item| item.parent_transaction_id == parent_txn_id) { d.event_id = Some(event_id.clone()); num_updated += 1; } @@ -1047,11 +1037,11 @@ impl StateStore for MemoryStore { async fn remove_dependent_send_queue_event( &self, room: &RoomId, - id: usize, + txn_id: &TransactionId, ) -> Result { let mut dependent_send_queue_events = self.dependent_send_queue_events.write().unwrap(); let dependents = dependent_send_queue_events.entry(room.to_owned()).or_default(); - if let Some(pos) = dependents.iter().position(|item| item.id == id) { + if let Some(pos) = dependents.iter().position(|item| item.own_transaction_id == txn_id) { dependents.remove(pos); Ok(true) } else { diff --git a/crates/matrix-sdk-base/src/store/traits.rs b/crates/matrix-sdk-base/src/store/traits.rs index ef83b9896..eac9d750b 100644 --- a/crates/matrix-sdk-base/src/store/traits.rs +++ b/crates/matrix-sdk-base/src/store/traits.rs @@ -459,7 +459,8 @@ pub trait StateStore: AsyncTraitDeps { async fn save_dependent_send_queue_event( &self, room_id: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, + own_txn_id: OwnedTransactionId, content: DependentQueuedEventKind, ) -> Result<(), Self::Error>; @@ -470,7 +471,7 @@ pub trait StateStore: AsyncTraitDeps { async fn update_dependent_send_queue_event( &self, room_id: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, event_id: OwnedEventId, ) -> Result; @@ -480,7 +481,7 @@ pub trait StateStore: AsyncTraitDeps { async fn remove_dependent_send_queue_event( &self, room: &RoomId, - id: usize, + own_txn_id: &TransactionId, ) -> Result; /// List all the dependent send queue events. @@ -767,11 +768,12 @@ impl StateStore for EraseStateStoreError { async fn save_dependent_send_queue_event( &self, room_id: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, + own_txn_id: OwnedTransactionId, content: DependentQueuedEventKind, ) -> Result<(), Self::Error> { self.0 - .save_dependent_send_queue_event(room_id, transaction_id, content) + .save_dependent_send_queue_event(room_id, parent_txn_id, own_txn_id, content) .await .map_err(Into::into) } @@ -779,11 +781,11 @@ impl StateStore for EraseStateStoreError { async fn update_dependent_send_queue_event( &self, room_id: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, event_id: OwnedEventId, ) -> Result { self.0 - .update_dependent_send_queue_event(room_id, transaction_id, event_id) + .update_dependent_send_queue_event(room_id, parent_txn_id, event_id) .await .map_err(Into::into) } @@ -791,9 +793,9 @@ impl StateStore for EraseStateStoreError { async fn remove_dependent_send_queue_event( &self, room_id: &RoomId, - id: usize, + own_txn_id: &TransactionId, ) -> Result { - self.0.remove_dependent_send_queue_event(room_id, id).await.map_err(Into::into) + self.0.remove_dependent_send_queue_event(room_id, own_txn_id).await.map_err(Into::into) } async fn list_dependent_send_queue_events( @@ -1262,7 +1264,7 @@ pub struct DependentQueuedEvent { /// Unique identifier for this dependent queued event. /// /// Useful for deletion. - pub id: usize, + pub own_transaction_id: OwnedTransactionId, /// The kind of user intent. pub kind: DependentQueuedEventKind, @@ -1272,7 +1274,7 @@ pub struct DependentQueuedEvent { /// Note: this is the transaction id used for the depended-on event, i.e. /// the one that was originally sent and that's being modified with this /// dependent event. - pub transaction_id: OwnedTransactionId, + pub parent_transaction_id: OwnedTransactionId, /// If the parent event has been sent, the parent's event identifier /// returned by the server once the local echo has been sent out. diff --git a/crates/matrix-sdk-indexeddb/src/state_store/mod.rs b/crates/matrix-sdk-indexeddb/src/state_store/mod.rs index 9054d04fa..e02cc4435 100644 --- a/crates/matrix-sdk-indexeddb/src/state_store/mod.rs +++ b/crates/matrix-sdk-indexeddb/src/state_store/mod.rs @@ -1575,7 +1575,8 @@ impl_state_store!({ async fn save_dependent_send_queue_event( &self, room_id: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, + own_txn_id: OwnedTransactionId, content: DependentQueuedEventKind, ) -> Result<()> { let encoded_key = self.encode_key(keys::DEPENDENT_SEND_QUEUE, room_id); @@ -1596,14 +1597,11 @@ impl_state_store!({ |val| self.deserialize_value::>(&val), )?; - // Find the next id by taking the biggest ID we had before, and add 1. - let next_id = prev.iter().fold(0, |max, item| item.id.max(max)) + 1; - // Push the new event. prev.push(DependentQueuedEvent { - id: next_id, kind: content, - transaction_id: transaction_id.to_owned(), + parent_transaction_id: parent_txn_id.to_owned(), + own_transaction_id: own_txn_id, event_id: None, }); @@ -1618,7 +1616,7 @@ impl_state_store!({ async fn update_dependent_send_queue_event( &self, room_id: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, event_id: OwnedEventId, ) -> Result { let encoded_key = self.encode_key(keys::DEPENDENT_SEND_QUEUE, room_id); @@ -1641,7 +1639,7 @@ impl_state_store!({ // Modify all events that match. let mut num_updated = 0; - for entry in prev.iter_mut().filter(|entry| entry.transaction_id == transaction_id) { + for entry in prev.iter_mut().filter(|entry| entry.parent_transaction_id == parent_txn_id) { entry.event_id = Some(event_id.clone()); num_updated += 1; } @@ -1654,7 +1652,11 @@ impl_state_store!({ Ok(num_updated) } - async fn remove_dependent_send_queue_event(&self, room_id: &RoomId, id: usize) -> Result { + async fn remove_dependent_send_queue_event( + &self, + room_id: &RoomId, + txn_id: &TransactionId, + ) -> Result { let encoded_key = self.encode_key(keys::DEPENDENT_SEND_QUEUE, room_id); let tx = self.inner.transaction_on_one_with_mode( @@ -1668,7 +1670,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)?; - if let Some(pos) = prev.iter().position(|item| item.id == id) { + if let Some(pos) = prev.iter().position(|item| item.own_transaction_id == txn_id) { prev.remove(pos); if prev.is_empty() { diff --git a/crates/matrix-sdk-sqlite/migrations/state_store/005_send_queue_dependent_events.sql b/crates/matrix-sdk-sqlite/migrations/state_store/005_send_queue_dependent_events.sql index fc9268801..bb522d2d0 100644 --- a/crates/matrix-sdk-sqlite/migrations/state_store/005_send_queue_dependent_events.sql +++ b/crates/matrix-sdk-sqlite/migrations/state_store/005_send_queue_dependent_events.sql @@ -4,7 +4,12 @@ CREATE TABLE "dependent_send_queue_events" ( "room_id" BLOB NOT NULL, -- This is used as both a key and a value, thus neither encrypted/decrypted/hashed. - "transaction_id" BLOB NOT NULL, + -- This is the transaction id for the *parent* transaction, not our own. + "parent_transaction_id" BLOB NOT NULL, + + -- This is used as both a key and a value, thus neither encrypted/decrypted/hashed. + -- This is a transaction id used for the dependent event itself, not the parent. + "own_transaction_id" BLOB NOT NULL, -- Used as a value (thus encrypted/decrypted), can be null. "event_id" BLOB NULL, diff --git a/crates/matrix-sdk-sqlite/src/state_store.rs b/crates/matrix-sdk-sqlite/src/state_store.rs index a7d5a6285..8f9524403 100644 --- a/crates/matrix-sdk-sqlite/src/state_store.rs +++ b/crates/matrix-sdk-sqlite/src/state_store.rs @@ -1862,19 +1862,26 @@ impl StateStore for SqliteStateStore { async fn save_dependent_send_queue_event( &self, room_id: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, + own_txn_id: OwnedTransactionId, content: DependentQueuedEventKind, ) -> Result<()> { let room_id = self.encode_key(keys::DEPENDENTS_SEND_QUEUE, room_id); let content = self.serialize_json(&content)?; // See comment in `save_send_queue_event`. - let transaction_id = transaction_id.to_string(); + let parent_txn_id = parent_txn_id.to_string(); + let own_txn_id = own_txn_id.to_string(); self.acquire() .await? .with_transaction(move |txn| { - txn.prepare_cached("INSERT INTO dependent_send_queue_events (room_id, transaction_id, content) VALUES (?, ?, ?)")?.execute((room_id, transaction_id, content))?; + txn.prepare_cached( + r#"INSERT INTO dependent_send_queue_events + (room_id, parent_transaction_id, own_transaction_id, content) + VALUES (?, ?, ?, ?)"#, + )? + .execute((room_id, parent_txn_id, own_txn_id, content))?; Ok(()) }) .await @@ -1883,37 +1890,44 @@ impl StateStore for SqliteStateStore { async fn update_dependent_send_queue_event( &self, room_id: &RoomId, - transaction_id: &TransactionId, + parent_txn_id: &TransactionId, event_id: OwnedEventId, ) -> Result { let room_id = self.encode_key(keys::DEPENDENTS_SEND_QUEUE, room_id); let event_id = self.serialize_value(&event_id)?; // See comment in `save_send_queue_event`. - let transaction_id = transaction_id.to_string(); + let parent_txn_id = parent_txn_id.to_string(); self.acquire() .await? .with_transaction(move |txn| { Ok(txn.prepare_cached( - "UPDATE dependent_send_queue_events SET event_id = ? WHERE transaction_id = ? and room_id = ?", + "UPDATE dependent_send_queue_events SET event_id = ? WHERE parent_transaction_id = ? and room_id = ?", )? - .execute((event_id, transaction_id, room_id))?) + .execute((event_id, parent_txn_id, room_id))?) }) .await } - async fn remove_dependent_send_queue_event(&self, room_id: &RoomId, id: usize) -> Result { + async fn remove_dependent_send_queue_event( + &self, + room_id: &RoomId, + txn_id: &TransactionId, + ) -> Result { let room_id = self.encode_key(keys::DEPENDENTS_SEND_QUEUE, room_id); + // See comment in `save_send_queue_event`. + let txn_id = txn_id.to_string(); + let num_deleted = self .acquire() .await? .with_transaction(move |txn| { txn.prepare_cached( - "DELETE FROM dependent_send_queue_events WHERE ROWID = ? AND room_id = ?", + "DELETE FROM dependent_send_queue_events WHERE own_transaction_id = ? AND room_id = ?", )? - .execute((id, room_id)) + .execute((txn_id, room_id)) }) .await?; @@ -1927,11 +1941,11 @@ impl StateStore for SqliteStateStore { let room_id = self.encode_key(keys::DEPENDENTS_SEND_QUEUE, room_id); // Note: transaction_id is not encoded, see why in `save_send_queue_event`. - let res: Vec<(usize, String, Option>, Vec)> = self + let res: Vec<(String, String, Option>, Vec)> = self .acquire() .await? .prepare( - "SELECT ROWID, transaction_id, event_id, content FROM dependent_send_queue_events WHERE room_id = ? ORDER BY ROWID", + "SELECT own_transaction_id, parent_transaction_id, event_id, content FROM dependent_send_queue_events WHERE room_id = ? ORDER BY ROWID", |mut stmt| { stmt.query((room_id,))? .mapped(|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))) @@ -1943,8 +1957,8 @@ impl StateStore for SqliteStateStore { let mut dependent_events = Vec::with_capacity(res.len()); for entry in res { dependent_events.push(DependentQueuedEvent { - id: entry.0, - transaction_id: entry.1.into(), + own_transaction_id: entry.0.into(), + parent_transaction_id: entry.1.into(), event_id: entry.2.map(|bytes| self.deserialize_value(&bytes)).transpose()?, kind: self.deserialize_json(&entry.3)?, }); diff --git a/crates/matrix-sdk/src/send_queue.rs b/crates/matrix-sdk/src/send_queue.rs index 5efb3e697..5759f7966 100644 --- a/crates/matrix-sdk/src/send_queue.rs +++ b/crates/matrix-sdk/src/send_queue.rs @@ -731,6 +731,7 @@ impl QueueStorage { .save_dependent_send_queue_event( &self.room_id, transaction_id, + TransactionId::new(), DependentQueuedEventKind::Redact, ) .await?; @@ -766,6 +767,7 @@ impl QueueStorage { .save_dependent_send_queue_event( &self.room_id, transaction_id, + TransactionId::new(), DependentQueuedEventKind::Edit { new_content: serializable }, ) .await?; @@ -856,7 +858,11 @@ impl QueueStorage { // The parent event is still local (sending must have failed); update the local // echo. let edited = store - .update_send_queue_event(&self.room_id, &de.transaction_id, new_content) + .update_send_queue_event( + &self.room_id, + &de.parent_transaction_id, + new_content, + ) .await .map_err(RoomSendQueueStorageError::StorageError)?; @@ -887,7 +893,7 @@ impl QueueStorage { // The parent event is still local (sending must have failed); redact the local // echo. let removed = store - .remove_send_queue_event(&self.room_id, &de.transaction_id) + .remove_send_queue_event(&self.room_id, &de.parent_transaction_id) .await .map_err(RoomSendQueueStorageError::StorageError)?; @@ -925,13 +931,13 @@ impl QueueStorage { ); for dependent in canonicalized_dependent_events { - let dependent_id = dependent.id; + let dependent_id = dependent.own_transaction_id.clone(); match self.try_apply_single_dependent_event(&client, dependent).await { Ok(()) => { // The dependent event has been successfully applied, forget about it. store - .remove_dependent_send_queue_event(&self.room_id, dependent_id) + .remove_dependent_send_queue_event(&self.room_id, &dependent_id) .await .map_err(RoomSendQueueStorageError::StorageError)?; @@ -1226,22 +1232,21 @@ mod tests { let txn = TransactionId::new(); let edit = DependentQueuedEvent { - id: 0, + own_transaction_id: TransactionId::new(), + parent_transaction_id: txn.clone(), kind: DependentQueuedEventKind::Edit { new_content: SerializableEventContent::new( &RoomMessageEventContent::text_plain("edit").into(), ) .unwrap(), }, - transaction_id: txn.clone(), event_id: None, }; let res = canonicalize_dependent_events(&[edit]); assert_eq!(res.len(), 1); - assert_eq!(res[0].id, 0); assert_matches!(&res[0].kind, DependentQueuedEventKind::Edit { .. }); - assert_eq!(res[0].transaction_id, txn); + assert_eq!(res[0].parent_transaction_id, txn); assert!(res[0].event_id.is_none()); } @@ -1252,35 +1257,35 @@ mod tests { let mut inputs = Vec::with_capacity(100); let redact = DependentQueuedEvent { - id: 0, + own_transaction_id: TransactionId::new(), + parent_transaction_id: txn.clone(), kind: DependentQueuedEventKind::Redact, - transaction_id: txn.clone(), event_id: None, }; let edit = DependentQueuedEvent { - id: 0, + own_transaction_id: TransactionId::new(), + parent_transaction_id: TransactionId::new(), kind: DependentQueuedEventKind::Edit { new_content: SerializableEventContent::new( &RoomMessageEventContent::text_plain("edit").into(), ) .unwrap(), }, - transaction_id: TransactionId::new(), event_id: None, }; inputs.push({ let mut edit = edit.clone(); - edit.id = 1; + edit.own_transaction_id = TransactionId::new(); edit }); inputs.push(redact); - for i in 0..98 { + for _ in 0..98 { let mut edit = edit.clone(); - edit.id = 2 + i; + edit.own_transaction_id = TransactionId::new(); inputs.push(edit); } @@ -1288,7 +1293,7 @@ mod tests { assert_eq!(res.len(), 1); assert_matches!(&res[0].kind, DependentQueuedEventKind::Redact); - assert_eq!(res[0].transaction_id, txn); + assert_eq!(res[0].parent_transaction_id, txn); } #[test] @@ -1296,19 +1301,19 @@ mod tests { // The latest edit of a list is always preferred. let inputs = (0..10) .map(|i| DependentQueuedEvent { - id: i, + own_transaction_id: TransactionId::new(), + parent_transaction_id: TransactionId::new(), kind: DependentQueuedEventKind::Edit { new_content: SerializableEventContent::new( &RoomMessageEventContent::text_plain(format!("edit{i}")).into(), ) .unwrap(), }, - transaction_id: TransactionId::new(), event_id: None, }) .collect::>(); - let txn = inputs[9].transaction_id.clone(); + let txn = inputs[9].parent_transaction_id.clone(); let res = canonicalize_dependent_events(&inputs); @@ -1318,6 +1323,6 @@ mod tests { AnyMessageLikeEventContent::RoomMessage(msg) = new_content.deserialize().unwrap() ); assert_eq!(msg.body(), "edit9"); - assert_eq!(res[0].transaction_id, txn); + assert_eq!(res[0].parent_transaction_id, txn); } }