fix: don't retry some relay errors (#370)

* fix: don't retry some relay errors * chore: comment
WalletConnect · Feb 22, 2024 · 7866bd5 · 7866bd5
1 parent 3e4896d
commit 7866bd5
Show file tree

Hide file tree

Showing 6 changed files with 78 additions and 13 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -72,8 +72,8 @@ futures = "0.3.26"
 futures-util = "0.3"
 dashmap = "5.4.0"
 
-relay_rpc = { git = "https://github.com/WalletConnect/WalletConnectRust.git", tag = "v0.26.4", features = ["cacao"] }
-relay_client = { git = "https://github.com/WalletConnect/WalletConnectRust.git", tag = "v0.26.4" }
+relay_rpc = { git = "https://github.com/WalletConnect/WalletConnectRust.git", branch = "feat/error-overhaul", features = ["cacao"] }
+relay_client = { git = "https://github.com/WalletConnect/WalletConnectRust.git", branch = "feat/error-overhaul" }
 x25519-dalek = { version = "2.0.0", features = ["static_secrets"] }
 hkdf = "0.12.3"
 sha2 = "0.10.6"

diff --git a/src/error.rs b/src/error.rs
@@ -9,9 +9,11 @@ use {
     chacha20poly1305::aead,
     data_encoding::DecodeError,
     hyper::StatusCode,
+    relay_client::error::ClientError,
     relay_rpc::{
         auth::{did::DidError, ed25519_dalek::ed25519},
         domain::{ClientIdDecodingError, ProjectId, Topic},
+        rpc::{PublishError, SubscriptionError, WatchError},
     },
     serde_json::json,
     std::{array::TryFromSliceError, string::FromUtf8Error, sync::Arc},
@@ -79,7 +81,16 @@ pub enum NotifyServerError {
     Base64Decode(#[from] base64::DecodeError),
 
     #[error(transparent)]
-    WalletConnectClient(#[from] relay_client::error::Error),
+    RelayClientError(#[from] ClientError),
+
+    #[error(transparent)]
+    RelaySubscribeError(#[from] relay_client::error::Error<SubscriptionError>),
+
+    #[error(transparent)]
+    RelayPublishError(#[from] relay_client::error::Error<PublishError>),
+
+    #[error(transparent)]
+    RelayWatchError(#[from] relay_client::error::Error<WatchError>),
 
     #[error(transparent)]
     TryRecvError(#[from] tokio::sync::mpsc::error::TryRecvError),

diff --git a/src/publish_relay_message.rs b/src/publish_relay_message.rs
@@ -6,7 +6,7 @@ use {
     relay_client::{error::Error, http::Client},
     relay_rpc::{
         domain::Topic,
-        rpc::{msg_id::get_message_id, Publish},
+        rpc::{self, msg_id::get_message_id, Publish, PublishError, SubscriptionError},
     },
     std::{
         sync::{Arc, OnceLock},
@@ -29,7 +29,7 @@ pub async fn publish_relay_message(
     relay_client: &Client,
     publish: &Publish,
     metrics: Option<&Metrics>,
-) -> Result<(), Error> {
+) -> Result<(), Error<PublishError>> {
     info!("publish_relay_message");
     let start = Instant::now();
 
@@ -47,7 +47,19 @@ pub async fn publish_relay_message(
         if let Some(metrics) = metrics {
             metrics.relay_outgoing_message_publish(publish.tag, start);
         }
-        result
+        match result {
+            Ok(_) => Ok(()),
+            Err(e) => match e {
+                Error::Response(rpc::Error::Handler(PublishError::MailboxLimitExceeded)) => {
+                    // Only happens if there is no subscriber for the topic in the first place.
+                    // So client is not expecting a response, or in the case of notify messages
+                    // should use getNotifications to get old notifications anyway.
+                    info!("Mailbox limit exceeded for topic {}", publish.topic);
+                    Ok(())
+                }
+                e => Err(e),
+            },
+        }
     };
 
     let mut tries = 0;
@@ -88,7 +100,7 @@ pub async fn subscribe_relay_topic(
     relay_client: &Client,
     topic: &Topic,
     metrics: Option<&Metrics>,
-) -> Result<(), Error> {
+) -> Result<(), Error<SubscriptionError>> {
     info!("subscribe_relay_topic");
     let start = Instant::now();
 
@@ -98,7 +110,20 @@ pub async fn subscribe_relay_topic(
         if let Some(metrics) = metrics {
             metrics.relay_subscribe_request(start);
         }
-        result
+        match result {
+            Ok(_) => Ok(()),
+            Err(e) => match e {
+                Error::Response(rpc::Error::Handler(
+                    SubscriptionError::SubscriberLimitExceeded,
+                )) => {
+                    // FIXME figure out how to handle this properly; being unable to subscribe means a broken state
+                    // https://walletconnect.slack.com/archives/C058RS0MH38/p1708183383748259
+                    warn!("Subscriber limit exceeded for topic {topic}");
+                    Ok(())
+                }
+                e => Err(e),
+            },
+        }
     };
 
     let mut tries = 0;
@@ -141,7 +166,7 @@ pub async fn extend_subscription_ttl(
     relay_client: &Client,
     topic: Topic,
     metrics: Option<&Metrics>,
-) -> Result<(), Error> {
+) -> Result<(), Error<PublishError>> {
     info!("extend_subscription_ttl");
 
     // Extremely minor performance optimization with OnceLock to avoid allocating the same empty string everytime

diff --git a/src/services/relay_renewal_job/refresh_topic_subscriptions.rs b/src/services/relay_renewal_job/refresh_topic_subscriptions.rs
@@ -71,10 +71,12 @@ pub async fn run(
                         // Subscribe a second time as the initial subscription above may have expired
                         subscribe_relay_topic(client, &topic, metrics)
                             .map_ok(|_| ())
+                            .map_err(NotifyServerError::from)
                             .and_then(|_| {
                                 // Subscribing only guarantees 5m TTL, so we always need to extend it.
                                 extend_subscription_ttl(client, topic.clone(), metrics)
                                     .map_ok(|_| ())
+                                    .map_err(Into::into)
                             })
                             .await
                     })

diff --git a/src/services/relay_renewal_job/register_webhook.rs b/src/services/relay_renewal_job/register_webhook.rs
@@ -6,15 +6,19 @@ use {
     },
     relay_rpc::{
         auth::ed25519_dalek::SigningKey,
-        rpc::{WatchStatus, WatchType},
+        rpc::{WatchError, WatchStatus, WatchType},
     },
     std::time::Duration,
     tracing::instrument,
     url::Url,
 };
 
 #[instrument(skip_all)]
-pub async fn run(notify_url: &Url, keypair: &SigningKey, client: &Client) -> Result<(), Error> {
+pub async fn run(
+    notify_url: &Url,
+    keypair: &SigningKey,
+    client: &Client,
+) -> Result<(), Error<WatchError>> {
     client
         .watch_register(
             WatchRegisterRequest {