send queue: flag unrecoverable errors, and don't block the queue on those

bnjbvr · bnjbvr · commit f45098a61d60 · 2024-06-17T17:15:07.000+02:00
diff --git a/bindings/matrix-sdk-ffi/src/timeline/mod.rs b/bindings/matrix-sdk-ffi/src/timeline/mod.rs
@@ -852,7 +852,16 @@ pub enum EventSendState {
     NotSentYet,
     /// The local event has been sent to the server, but unsuccessfully: The
     /// sending has failed.
-    SendingFailed { error: String },
+    SendingFailed {
+        /// Stringified error message.
+        error: String,
+        /// Whether the error is considered recoverable or not.
+        ///
+        /// An error that's recoverable will disable the room's send queue,
+        /// while an unrecoverable error will be parked, until the user
+        /// decides to cancel sending it.
+        is_recoverable: bool,
+    },
     /// The local event has been sent successfully to the server.
     Sent { event_id: String },
 }
@@ -863,7 +872,9 @@ impl From<&matrix_sdk_ui::timeline::EventSendState> for EventSendState {
 
         match value {
             NotSentYet => Self::NotSentYet,
-            SendingFailed { error } => Self::SendingFailed { error: error.to_string() },
+            SendingFailed { error, is_recoverable } => {
+                Self::SendingFailed { error: error.to_string(), is_recoverable: *is_recoverable }
+            }
             Sent { event_id } => Self::Sent { event_id: event_id.to_string() },
         }
     }
diff --git a/crates/matrix-sdk-ui/src/timeline/builder.rs b/crates/matrix-sdk-ui/src/timeline/builder.rs
@@ -318,11 +318,15 @@ impl TimelineBuilder {
                                     }
                                 }
 
-                                RoomSendQueueUpdate::SendError { transaction_id, error } => {
+                                RoomSendQueueUpdate::SendError {
+                                    transaction_id,
+                                    error,
+                                    is_recoverable,
+                                } => {
                                     timeline
                                         .update_event_send_state(
                                             &transaction_id,
-                                            EventSendState::SendingFailed { error },
+                                            EventSendState::SendingFailed { error, is_recoverable },
                                         )
                                         .await;
                                 }
diff --git a/crates/matrix-sdk-ui/src/timeline/event_item/local.rs b/crates/matrix-sdk-ui/src/timeline/event_item/local.rs
@@ -55,6 +55,12 @@ pub enum EventSendState {
     SendingFailed {
         /// Details about how sending the event failed.
         error: Arc<Error>,
+        /// Whether the error is considered recoverable or not.
+        ///
+        /// An error that's recoverable will disable the room's send queue,
+        /// while an unrecoverable error will be parked, until the user
+        /// decides to cancel sending it.
+        is_recoverable: bool,
     },
     /// The local event has been sent successfully to the server.
     Sent {
diff --git a/crates/matrix-sdk-ui/src/timeline/tests/echo.rs b/crates/matrix-sdk-ui/src/timeline/tests/echo.rs
@@ -63,14 +63,20 @@ async fn test_remote_echo_full_trip() {
             .inner
             .update_event_send_state(
                 &txn_id,
-                EventSendState::SendingFailed { error: Arc::new(some_io_error) },
+                EventSendState::SendingFailed {
+                    error: Arc::new(some_io_error),
+                    is_recoverable: true,
+                },
             )
             .await;
 
         let item = assert_next_matches!(stream, VectorDiff::Set { value, index: 1 } => value);
         let event_item = item.as_event().unwrap();
         assert!(event_item.is_local_echo());
-        assert_matches!(event_item.send_state(), Some(EventSendState::SendingFailed { .. }));
+        assert_matches!(
+            event_item.send_state(),
+            Some(EventSendState::SendingFailed { is_recoverable: true, .. })
+        );
         assert_eq!(item.unique_id(), id);
     }
 
diff --git a/crates/matrix-sdk-ui/tests/integration/timeline/echo.rs b/crates/matrix-sdk-ui/tests/integration/timeline/echo.rs
@@ -138,9 +138,6 @@ async fn test_retry_failed() {
 
     mock_sync(&server, sync_builder.build_json_sync_response(), None).await;
     let _response = client.sync_once(sync_settings.clone()).await.unwrap();
-    server.reset().await;
-
-    mock_encryption_state(&server, false).await;
 
     client.send_queue().set_enabled(true);
 
@@ -149,20 +146,43 @@ async fn test_retry_failed() {
     let (_, mut timeline_stream) =
         timeline.subscribe_filter_map(|item| item.as_event().cloned()).await;
 
+    // When trying to send an event, return with a 500 error, which is interpreted
+    // as a transient error.
+    server.reset().await;
+    mock_encryption_state(&server, false).await;
+    let scoped_faulty_send = Mock::given(method("PUT"))
+        .and(path_regex(r"^/_matrix/client/r0/rooms/.*/send/.*"))
+        .and(header("authorization", "Bearer 1234"))
+        .respond_with(ResponseTemplate::new(500))
+        .expect(3)
+        .mount_as_scoped(&server)
+        .await;
+
     timeline.send(RoomMessageEventContent::text_plain("Hello, World!").into()).await.unwrap();
 
     // Let the send queue handle the event.
     yield_now().await;
 
-    // First, local echo is added
+    // First, local echo is added.
     assert_next_matches!(timeline_stream, VectorDiff::PushBack { value } => {
         assert_matches!(value.send_state(), Some(EventSendState::NotSentYet));
     });
 
-    // Sending fails, the mock server has no matching route yet
+    // Sending fails, because the error is a transient one that's recoverable,
+    // indicating something's wrong on the client side.
     assert_let!(Some(VectorDiff::Set { index: 0, value: item }) = timeline_stream.next().await);
-    assert_matches!(item.send_state(), Some(EventSendState::SendingFailed { .. }));
+    assert_matches!(
+        item.send_state(),
+        Some(EventSendState::SendingFailed { is_recoverable: true, .. })
+    );
 
+    // This doesn't disable the send queue at the global level…
+    assert!(client.send_queue().is_enabled());
+    // …but does so at the local level.
+    assert!(!room.send_queue().is_enabled());
+
+    // Have the endpoint return a success result, and re-enable the queue.
+    drop(scoped_faulty_send);
     Mock::given(method("PUT"))
         .and(path_regex(r"^/_matrix/client/r0/rooms/.*/send/.*"))
         .and(header("authorization", "Bearer 1234"))
@@ -173,11 +193,6 @@ async fn test_retry_failed() {
         .mount(&server)
         .await;
 
-    // This doesn't disable the send queue at the global level…
-    assert!(client.send_queue().is_enabled());
-    // …but does so at the local level.
-    assert!(!room.send_queue().is_enabled());
-
     room.send_queue().set_enabled(true);
 
     // Let the send queue handle the event.
diff --git a/crates/matrix-sdk-ui/tests/integration/timeline/queue.rs b/crates/matrix-sdk-ui/tests/integration/timeline/queue.rs
@@ -28,7 +28,7 @@ use serde_json::json;
 use stream_assert::{assert_next_matches, assert_pending};
 use tokio::{task::yield_now, time::sleep};
 use wiremock::{
-    matchers::{body_string_contains, method, path_regex},
+    matchers::{body_string_contains, header, method, path_regex},
     Mock, ResponseTemplate,
 };
 
@@ -126,17 +126,26 @@ async fn test_retry_order() {
 
     mock_sync(&server, sync_response_builder.build_json_sync_response(), None).await;
     let _response = client.sync_once(sync_settings.clone()).await.unwrap();
-    server.reset().await;
-
-    mock_encryption_state(&server, false).await;
 
     let room = client.get_room(room_id).unwrap();
     let timeline = Arc::new(room.timeline().await.unwrap());
     let (_, mut timeline_stream) =
         timeline.subscribe_filter_map(|item| item.as_event().cloned()).await;
 
+    // When trying to send an event, return with a 500 error, which is interpreted
+    // as a transient error.
+    server.reset().await;
+    mock_encryption_state(&server, false).await;
+    let scoped_faulty_send = Mock::given(method("PUT"))
+        .and(path_regex(r"^/_matrix/client/r0/rooms/.*/send/.*"))
+        .and(header("authorization", "Bearer 1234"))
+        .respond_with(ResponseTemplate::new(500))
+        .expect(3)
+        .mount_as_scoped(&server)
+        .await;
+
     // Send two messages without mocking the server response.
-    // It will respond with a 404, resulting in a failed-to-send state.
+    // It will respond with a 500, resulting in a failed-to-send state.
     timeline.send(RoomMessageEventContent::text_plain("First!").into()).await.unwrap();
     timeline.send(RoomMessageEventContent::text_plain("Second.").into()).await.unwrap();
 
@@ -157,6 +166,7 @@ async fn test_retry_order() {
     assert_matches!(first.send_state().unwrap(), EventSendState::SendingFailed { .. });
 
     // Response for first message takes 100ms to respond
+    drop(scoped_faulty_send);
     Mock::given(method("PUT"))
         .and(path_regex(r"^/_matrix/client/r0/rooms/.*/send/.*"))
         .and(body_string_contains("First!"))
diff --git a/crates/matrix-sdk/src/http_client/mod.rs b/crates/matrix-sdk/src/http_client/mod.rs
@@ -259,3 +259,53 @@ impl tower::Service<http_old::Request<Bytes>> for HttpClient {
         Box::pin(fut)
     }
 }
+
+pub(crate) enum RetryKind {
+    Transient {
+        err: HttpError,
+        // This is used only for attempts to retry, so on non-wasm32 code (in the `native` module).
+        #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
+        retry_after: Option<Duration>,
+    },
+    Permanent(HttpError),
+}
+
+impl RetryKind {
+    pub fn error(self) -> HttpError {
+        match self {
+            RetryKind::Transient { err, .. } | RetryKind::Permanent(err) => err,
+        }
+    }
+}
+
+/// Returns whether an API error is transient or permanent,
+pub(crate) fn characterize_retry_kind(err: HttpError) -> RetryKind {
+    use ruma::api::client::error::{ErrorBody, ErrorKind, RetryAfter};
+
+    use crate::RumaApiError;
+
+    if let Some(api_error) = err.as_ruma_api_error() {
+        let status_code = match api_error {
+            RumaApiError::ClientApi(e) => match e.body {
+                ErrorBody::Standard { kind: ErrorKind::LimitExceeded { retry_after }, .. } => {
+                    let retry_after = retry_after.and_then(|retry_after| match retry_after {
+                        RetryAfter::Delay(d) => Some(d),
+                        RetryAfter::DateTime(_) => None,
+                    });
+                    return RetryKind::Transient { err, retry_after };
+                }
+                _ => Some(e.status_code),
+            },
+            RumaApiError::Uiaa(_) => None,
+            RumaApiError::Other(e) => Some(e.status_code),
+        };
+
+        if let Some(status_code) = status_code {
+            if status_code.is_server_error() {
+                return RetryKind::Transient { err, retry_after: None };
+            }
+        }
+    }
+
+    RetryKind::Permanent(err)
+}
diff --git a/crates/matrix-sdk/src/http_client/native.rs b/crates/matrix-sdk/src/http_client/native.rs
@@ -25,15 +25,14 @@ use bytesize::ByteSize;
 use eyeball::SharedObservable;
 use http::header::CONTENT_LENGTH;
 use reqwest::Certificate;
-use ruma::api::{
-    client::error::{ErrorBody as ClientApiErrorBody, ErrorKind as ClientApiErrorKind, RetryAfter},
-    error::FromHttpResponseError,
-    IncomingResponse, OutgoingRequest,
-};
+use ruma::api::{error::FromHttpResponseError, IncomingResponse, OutgoingRequest};
 use tracing::{info, warn};
 
-use super::{response_to_http_response, HttpClient, TransmissionProgress, DEFAULT_REQUEST_TIMEOUT};
-use crate::{config::RequestConfig, error::HttpError, RumaApiError};
+use super::{
+    characterize_retry_kind, response_to_http_response, HttpClient, RetryKind,
+    TransmissionProgress, DEFAULT_REQUEST_TIMEOUT,
+};
+use crate::{config::RequestConfig, error::HttpError};
 
 impl HttpClient {
     pub(super) async fn send_request<R>(
@@ -63,35 +62,11 @@ impl HttpClient {
                 let error_type = if stop {
                     RetryError::Permanent
                 } else {
-                    |err: HttpError| {
-                        if let Some(api_error) = err.as_ruma_api_error() {
-                            let status_code = match api_error {
-                                RumaApiError::ClientApi(e) => match e.body {
-                                    ClientApiErrorBody::Standard {
-                                        kind: ClientApiErrorKind::LimitExceeded { retry_after },
-                                        ..
-                                    } => {
-                                        let retry_after =
-                                            retry_after.and_then(|retry_after| match retry_after {
-                                                RetryAfter::Delay(d) => Some(d),
-                                                RetryAfter::DateTime(_) => None,
-                                            });
-                                        return RetryError::Transient { err, retry_after };
-                                    }
-                                    _ => Some(e.status_code),
-                                },
-                                RumaApiError::Uiaa(_) => None,
-                                RumaApiError::Other(e) => Some(e.status_code),
-                            };
-
-                            if let Some(status_code) = status_code {
-                                if status_code.is_server_error() {
-                                    return RetryError::Transient { err, retry_after: None };
-                                }
-                            }
+                    |err: HttpError| match characterize_retry_kind(err) {
+                        RetryKind::Transient { err, retry_after } => {
+                            RetryError::Transient { err, retry_after }
                         }
-
-                        RetryError::Permanent(err)
+                        RetryKind::Permanent(err) => RetryError::Permanent(err),
                     }
                 };
 
diff --git a/crates/matrix-sdk/src/send_queue.rs b/crates/matrix-sdk/src/send_queue.rs
diff --git a/crates/matrix-sdk/tests/integration/send_queue.rs b/crates/matrix-sdk/tests/integration/send_queue.rs

Original file line number	Diff line number	Diff line change
`@@ -318,11 +318,15 @@ impl TimelineBuilder {`
`318`	`318`	`}`
`319`	`319`	`}`
`320`	`320`
`321`		`- RoomSendQueueUpdate::SendError { transaction_id, error } => {`
	`321`	`+ RoomSendQueueUpdate::SendError {`
	`322`	`+ transaction_id,`
	`323`	`+ error,`
	`324`	`+ is_recoverable,`
	`325`	`+ } => {`
`322`	`326`	`timeline`
`323`	`327`	`.update_event_send_state(`
`324`	`328`	`&transaction_id,`
`325`		`- EventSendState::SendingFailed { error },`
	`329`	`+ EventSendState::SendingFailed { error, is_recoverable },`
`326`	`330`	`)`
`327`	`331`	`.await;`
`328`	`332`	`}`