Dima/matrix-rust-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

send queue: add an own transaction id for dependent events

Browse Source 
The previously named `transaction_id` is also renamed to
`parent_transaction_id` to make it clearer.
This commit is contained in:
Benjamin Bouvier
2024-07-24 17:16:47 +02:00
committed by Benjamin Bouvier
parent 0246863af3
commit 2eb6930988
7 changed files with 119 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/matrix-sdk-base/src/store/integration_tests.rs
+26 -10
View File
@@ -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(),
crates/matrix-sdk-base/src/store/memory_store.rs
+9 -19
View File
@@ -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<HashMap<Vec<u8>, Vec<u8>>>,
send_queue_events: StdRwLock<BTreeMap<OwnedRoomId, Vec<QueuedEvent>>>,
dependent_send_queue_events: StdRwLock<BTreeMap<OwnedRoomId, Vec<DependentQueuedEvent>>>,
dependent_send_queue_event_next_id: Mutex<usize>,
}
// 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<usize, Self::Error> {
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<bool, Self::Error> {
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 {
crates/matrix-sdk-base/src/store/traits.rs
+13 -11
View File
@@ -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<usize, Self::Error>;
@@ -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<bool, Self::Error>;
/// List all the dependent send queue events.
@@ -767,11 +768,12 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
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<T: StateStore> StateStore for EraseStateStoreError<T> {
async fn update_dependent_send_queue_event(
&self,
room_id: &RoomId,
transaction_id: &TransactionId,
parent_txn_id: &TransactionId,
event_id: OwnedEventId,
) -> Result<usize, Self::Error> {
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<T: StateStore> StateStore for EraseStateStoreError<T> {
async fn remove_dependent_send_queue_event(
&self,
room_id: &RoomId,
id: usize,
own_txn_id: &TransactionId,
) -> Result<bool, Self::Error> {
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.
crates/matrix-sdk-indexeddb/src/state_store/mod.rs
+12 -10
View File
@@ -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::<Vec<DependentQueuedEvent>>(&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<usize> {
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<bool> {
async fn remove_dependent_send_queue_event(
&self,
room_id: &RoomId,
txn_id: &TransactionId,
) -> Result<bool> {
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::<Vec<DependentQueuedEvent>>(&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() {
crates/matrix-sdk-sqlite/migrations/state_store/005_send_queue_dependent_events.sql
+6 -1
View File
@@ -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,
crates/matrix-sdk-sqlite/src/state_store.rs
+28 -14
View File
@@ -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<usize> {
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<bool> {
async fn remove_dependent_send_queue_event(
&self,
room_id: &RoomId,
txn_id: &TransactionId,
) -> Result<bool> {
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<u8>>, Vec<u8>)> = self
let res: Vec<(String, String, Option<Vec<u8>>, Vec<u8>)> = 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)?,
});
crates/matrix-sdk/src/send_queue.rs
+25 -20
View File
@@ -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::<Vec<_>>();
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);
}
}