feat(send queue): Add an enqueued time to to-be-sent events (#4385)

Add a new created_at to the send_queue_events and
dependent_send_queue_events stored records. This will allow clients to
understand how stale a pending message might be in the event that the
queue encounters and error and becomes wedged.

This change is exposed through the FFI on the `EventTimelineItem` struct
as a new optional field named `local_created_at`. It will be `None` for
any Remote event, and `Some` for Local events (except for those that
were enqueued before the migrations were run).

Signed-off-by: Daniel Salinas

---------

Signed-off-by: Daniel Salinas <[email protected]>
Co-authored-by: Daniel Salinas <[email protected]>
Co-authored-by: Benjamin Bouvier <[email protected]>
Co-authored-by: Daniel Salinas <[email protected]>

Author: 4 people

bindings/matrix-sdk-ffi/src/timeline/mod.rs

@@ -1033,6 +1033,7 @@ pub struct EventTimelineItem {
     timestamp: Timestamp,
     reactions: Vec<Reaction>,
     local_send_state: Option<EventSendState>,
+    local_created_at: Option<u64>,
     read_receipts: HashMap<String, Receipt>,
     origin: Option<EventItemOrigin>,
     can_be_replied_to: bool,
@@ -1070,6 +1071,7 @@ impl From<matrix_sdk_ui::timeline::EventTimelineItem> for EventTimelineItem {
             timestamp: item.timestamp().into(),
             reactions,
             local_send_state: item.send_state().map(|s| s.into()),
+            local_created_at: item.local_created_at().map(|t| t.0.into()),
             read_receipts,
             origin: item.origin(),
             can_be_replied_to: item.can_be_replied_to(),

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

@@ -29,7 +29,8 @@ use ruma::{
     },
     owned_event_id, owned_mxc_uri, room_id,
     serde::Raw,
-    uint, user_id, EventId, OwnedEventId, OwnedUserId, RoomId, TransactionId, UserId,
+    uint, user_id, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedUserId, RoomId,
+    TransactionId, UserId,
 };
 use serde_json::{json, value::Value as JsonValue};
 
@@ -980,13 +981,21 @@ impl StateStoreIntegrationTests for DynStateStore {
             let ev =
                 SerializableEventContent::new(&RoomMessageEventContent::text_plain("sup").into())
                     .unwrap();
-            self.save_send_queue_request(room_id, txn.clone(), ev.into(), 0).await?;
+            self.save_send_queue_request(
+                room_id,
+                txn.clone(),
+                MilliSecondsSinceUnixEpoch::now(),
+                ev.into(),
+                0,
+            )
+            .await?;
 
             // Add a single dependent queue request.
             self.save_dependent_queued_request(
                 room_id,
                 &txn,
                 ChildTransactionId::new(),
+                MilliSecondsSinceUnixEpoch::now(),
                 DependentQueuedRequestKind::RedactEvent,
             )
             .await?;
@@ -1242,7 +1251,15 @@ impl StateStoreIntegrationTests for DynStateStore {
         let event0 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("msg0").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, txn0.clone(), event0.into(), 0).await.unwrap();
+        self.save_send_queue_request(
+            room_id,
+            txn0.clone(),
+            MilliSecondsSinceUnixEpoch::now(),
+            event0.into(),
+            0,
+        )
+        .await
+        .unwrap();
 
         // Reading it will work.
         let pending = self.load_send_queue_requests(room_id).await.unwrap();
@@ -1266,7 +1283,15 @@ impl StateStoreIntegrationTests for DynStateStore {
             )
             .unwrap();
 
-            self.save_send_queue_request(room_id, txn, event.into(), 0).await.unwrap();
+            self.save_send_queue_request(
+                room_id,
+                txn,
+                MilliSecondsSinceUnixEpoch::now(),
+                event.into(),
+                0,
+            )
+            .await
+            .unwrap();
         }
 
         // Reading all the events should work.
@@ -1364,7 +1389,15 @@ impl StateStoreIntegrationTests for DynStateStore {
             let event =
                 SerializableEventContent::new(&RoomMessageEventContent::text_plain("room2").into())
                     .unwrap();
-            self.save_send_queue_request(room_id2, txn.clone(), event.into(), 0).await.unwrap();
+            self.save_send_queue_request(
+                room_id2,
+                txn.clone(),
+                MilliSecondsSinceUnixEpoch::now(),
+                event.into(),
+                0,
+            )
+            .await
+            .unwrap();
         }
 
         // Add and remove one event for room3.
@@ -1374,7 +1407,15 @@ impl StateStoreIntegrationTests for DynStateStore {
             let event =
                 SerializableEventContent::new(&RoomMessageEventContent::text_plain("room3").into())
                     .unwrap();
-            self.save_send_queue_request(room_id3, txn.clone(), event.into(), 0).await.unwrap();
+            self.save_send_queue_request(
+                room_id3,
+                txn.clone(),
+                MilliSecondsSinceUnixEpoch::now(),
+                event.into(),
+                0,
+            )
+            .await
+            .unwrap();
 
             self.remove_send_queue_request(room_id3, &txn).await.unwrap();
         }
@@ -1399,21 +1440,45 @@ impl StateStoreIntegrationTests for DynStateStore {
         let ev0 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("low0").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, low0_txn.clone(), ev0.into(), 2).await.unwrap();
+        self.save_send_queue_request(
+            room_id,
+            low0_txn.clone(),
+            MilliSecondsSinceUnixEpoch::now(),
+            ev0.into(),
+            2,
+        )
+        .await
+        .unwrap();
 
         // Saving one request with higher priority should work.
         let high_txn = TransactionId::new();
         let ev1 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("high").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, high_txn.clone(), ev1.into(), 10).await.unwrap();
+        self.save_send_queue_request(
+            room_id,
+            high_txn.clone(),
+            MilliSecondsSinceUnixEpoch::now(),
+            ev1.into(),
+            10,
+        )
+        .await
+        .unwrap();
 
         // Saving another request with the low priority should work.
         let low1_txn = TransactionId::new();
         let ev2 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("low1").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, low1_txn.clone(), ev2.into(), 2).await.unwrap();
+        self.save_send_queue_request(
+            room_id,
+            low1_txn.clone(),
+            MilliSecondsSinceUnixEpoch::now(),
+            ev2.into(),
+            2,
+        )
+        .await
+        .unwrap();
 
         // The requests should be ordered from higher priority to lower, and when equal,
         // should use the insertion order instead.
@@ -1453,7 +1518,15 @@ impl StateStoreIntegrationTests for DynStateStore {
         let event0 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("hey").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, txn0.clone(), event0.into(), 0).await.unwrap();
+        self.save_send_queue_request(
+            room_id,
+            txn0.clone(),
+            MilliSecondsSinceUnixEpoch::now(),
+            event0.into(),
+            0,
+        )
+        .await
+        .unwrap();
 
         // No dependents, to start with.
         assert!(self.load_dependent_queued_requests(room_id).await.unwrap().is_empty());
@@ -1464,6 +1537,7 @@ impl StateStoreIntegrationTests for DynStateStore {
             room_id,
             &txn0,
             child_txn.clone(),
+            MilliSecondsSinceUnixEpoch::now(),
             DependentQueuedRequestKind::RedactEvent,
         )
         .await
@@ -1515,12 +1589,21 @@ impl StateStoreIntegrationTests for DynStateStore {
         let event1 =
             SerializableEventContent::new(&RoomMessageEventContent::text_plain("hey2").into())
                 .unwrap();
-        self.save_send_queue_request(room_id, txn1.clone(), event1.into(), 0).await.unwrap();
+        self.save_send_queue_request(
+            room_id,
+            txn1.clone(),
+            MilliSecondsSinceUnixEpoch::now(),
+            event1.into(),
+            0,
+        )
+        .await
+        .unwrap();
 
         self.save_dependent_queued_request(
             room_id,
             &txn0,
             ChildTransactionId::new(),
+            MilliSecondsSinceUnixEpoch::now(),
             DependentQueuedRequestKind::RedactEvent,
         )
         .await
@@ -1531,6 +1614,7 @@ impl StateStoreIntegrationTests for DynStateStore {
             room_id,
             &txn1,
             ChildTransactionId::new(),
+            MilliSecondsSinceUnixEpoch::now(),
             DependentQueuedRequestKind::EditEvent {
                 new_content: SerializableEventContent::new(
                     &RoomMessageEventContent::text_plain("edit").into(),
@@ -1563,6 +1647,7 @@ impl StateStoreIntegrationTests for DynStateStore {
             room_id,
             &txn,
             child_txn.clone(),
+            MilliSecondsSinceUnixEpoch::now(),
             DependentQueuedRequestKind::RedactEvent,
         )
         .await

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

@@ -30,8 +30,8 @@ use ruma::{
     },
     serde::Raw,
     time::Instant,
-    CanonicalJsonObject, EventId, OwnedEventId, OwnedMxcUri, OwnedRoomId, OwnedTransactionId,
-    OwnedUserId, RoomId, RoomVersionId, TransactionId, UserId,
+    CanonicalJsonObject, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedMxcUri,
+    OwnedRoomId, OwnedTransactionId, OwnedUserId, RoomId, RoomVersionId, TransactionId, UserId,
 };
 use tracing::{debug, instrument, warn};
 
@@ -750,6 +750,7 @@ impl StateStore for MemoryStore {
         &self,
         room_id: &RoomId,
         transaction_id: OwnedTransactionId,
+        created_at: MilliSecondsSinceUnixEpoch,
         kind: QueuedRequestKind,
         priority: usize,
     ) -> Result<(), Self::Error> {
@@ -759,7 +760,7 @@ impl StateStore for MemoryStore {
             .send_queue_events
             .entry(room_id.to_owned())
             .or_default()
-            .push(QueuedRequest { kind, transaction_id, error: None, priority });
+            .push(QueuedRequest { kind, transaction_id, error: None, priority, created_at });
         Ok(())
     }
 
@@ -858,6 +859,7 @@ impl StateStore for MemoryStore {
         room: &RoomId,
         parent_transaction_id: &TransactionId,
         own_transaction_id: ChildTransactionId,
+        created_at: MilliSecondsSinceUnixEpoch,
         content: DependentQueuedRequestKind,
     ) -> Result<(), Self::Error> {
         self.inner
@@ -871,6 +873,7 @@ impl StateStore for MemoryStore {
                 parent_transaction_id: parent_transaction_id.to_owned(),
                 own_transaction_id,
                 parent_key: None,
+                created_at,
             });
         Ok(())
     }

crates/matrix-sdk-base/src/store/send_queue.rs

@@ -23,7 +23,8 @@ use ruma::{
         AnyMessageLikeEventContent, EventContent as _, RawExt as _,
     },
     serde::Raw,
-    OwnedDeviceId, OwnedEventId, OwnedTransactionId, OwnedUserId, TransactionId, UInt,
+    MilliSecondsSinceUnixEpoch, OwnedDeviceId, OwnedEventId, OwnedTransactionId, OwnedUserId,
+    TransactionId, UInt,
 };
 use serde::{Deserialize, Serialize};
 
@@ -131,6 +132,9 @@ pub struct QueuedRequest {
     /// The bigger the value, the higher the priority at which this request
     /// should be handled.
     pub priority: usize,
+
+    /// The time that the request was originally attempted.
+    pub created_at: MilliSecondsSinceUnixEpoch,
 }
 
 impl QueuedRequest {
@@ -371,6 +375,9 @@ pub struct DependentQueuedRequest {
     /// If the parent request has been sent, the parent's request identifier
     /// returned by the server once the local echo has been sent out.
     pub parent_key: Option<SentRequestKey>,
+
+    /// The time that the request was originally attempted.
+    pub created_at: MilliSecondsSinceUnixEpoch,
 }
 
 impl DependentQueuedRequest {

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

@@ -35,8 +35,8 @@ use ruma::{
     },
     serde::Raw,
     time::SystemTime,
-    EventId, OwnedEventId, OwnedMxcUri, OwnedRoomId, OwnedTransactionId, OwnedUserId, RoomId,
-    TransactionId, UserId,
+    EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedMxcUri, OwnedRoomId,
+    OwnedTransactionId, OwnedUserId, RoomId, TransactionId, UserId,
 };
 use serde::{Deserialize, Serialize};
 
@@ -359,6 +359,7 @@ pub trait StateStore: AsyncTraitDeps {
         &self,
         room_id: &RoomId,
         transaction_id: OwnedTransactionId,
+        created_at: MilliSecondsSinceUnixEpoch,
         request: QueuedRequestKind,
         priority: usize,
     ) -> Result<(), Self::Error>;
@@ -421,6 +422,7 @@ pub trait StateStore: AsyncTraitDeps {
         room_id: &RoomId,
         parent_txn_id: &TransactionId,
         own_txn_id: ChildTransactionId,
+        created_at: MilliSecondsSinceUnixEpoch,
         content: DependentQueuedRequestKind,
     ) -> Result<(), Self::Error>;
 
@@ -657,11 +659,12 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
         &self,
         room_id: &RoomId,
         transaction_id: OwnedTransactionId,
+        created_at: MilliSecondsSinceUnixEpoch,
         content: QueuedRequestKind,
         priority: usize,
     ) -> Result<(), Self::Error> {
         self.0
-            .save_send_queue_request(room_id, transaction_id, content, priority)
+            .save_send_queue_request(room_id, transaction_id, created_at, content, priority)
             .await
             .map_err(Into::into)
     }
@@ -711,10 +714,11 @@ impl<T: StateStore> StateStore for EraseStateStoreError<T> {
         room_id: &RoomId,
         parent_txn_id: &TransactionId,
         own_txn_id: ChildTransactionId,
+        created_at: MilliSecondsSinceUnixEpoch,
         content: DependentQueuedRequestKind,
     ) -> Result<(), Self::Error> {
         self.0
-            .save_dependent_queued_request(room_id, parent_txn_id, own_txn_id, content)
+            .save_dependent_queued_request(room_id, parent_txn_id, own_txn_id, created_at, content)
             .await
             .map_err(Into::into)
     }