add retry possibilities to redap client and retry on connection establishment

### What

Adding a simple and conservative retry logic to redap client. Using it on the
connection establishment path for now, only of native, as this is the
most common error we saw during concurrent readers testing.

Code is mostly copy-paste from chunk scanner client in dataplatform.
``re_backoff`` is a clone of ``redap_backoff``. Can start using it on
the dataplatform side after I merge this.

Source-Ref: dc97f1937db58b3c84f10c4faa3b84bb8d1d4398

Author: zehiko

ARCHITECTURE.md

@@ -225,6 +225,7 @@ Update instructions:
 | re_arrow_combinators | Type-safe, composable transformations for Arrow arrays                               |
 | re_arrow_util        | Helpers for working with arrow                                                       |
 | re_auth              | Authentication and authorization helpers                                             |
+| re_backoff           | Simple backoff logic used for retries.                                               |
 | re_byte_size         | Calculate the heap-allocated size of values at runtime                               |
 | re_capabilities      | Capability tokens                                                                    |
 | re_case              | Case conversions, the way Rerun likes them                                           |

Cargo.lock

@@ -8429,6 +8429,14 @@ dependencies = [
  "webbrowser",
 ]
 
+[[package]]
+name = "re_backoff"
+version = "0.30.0-alpha.1+dev"
+dependencies = [
+ "rand 0.9.2",
+ "tokio",
+]
+
 [[package]]
 name = "re_blueprint_tree"
 version = "0.30.0-alpha.1+dev"
@@ -9454,6 +9462,7 @@ dependencies = [
  "jiff",
  "re_arrow_util",
  "re_auth",
+ "re_backoff",
  "re_byte_size",
  "re_chunk",
  "re_format",

Cargo.toml

@@ -100,6 +100,7 @@ rerun-cli = { path = "crates/top/rerun-cli", version = "=0.30.0-alpha.1", defaul
 re_analytics = { path = "crates/utils/re_analytics", version = "=0.30.0-alpha.1", default-features = false }
 re_arrow_util = { path = "crates/utils/re_arrow_util", version = "=0.30.0-alpha.1", default-features = false }
 re_auth = { path = "crates/utils/re_auth", version = "=0.30.0-alpha.1", default-features = false }
+re_backoff = { path = "crates/utils/re_backoff", version = "=0.30.0-alpha.1", default-features = false }
 re_byte_size = { path = "crates/utils/re_byte_size", version = "=0.30.0-alpha.1", default-features = false }
 re_capabilities = { path = "crates/utils/re_capabilities", version = "=0.30.0-alpha.1", default-features = false }
 re_case = { path = "crates/utils/re_case", version = "=0.30.0-alpha.1", default-features = false }

crates/store/re_redap_client/Cargo.toml

@@ -33,6 +33,7 @@ all-features = true
 [dependencies]
 re_arrow_util.workspace = true
 re_auth = { workspace = true, features = ["oauth"] }
+re_backoff.workspace = true
 re_byte_size.workspace = true
 re_chunk.workspace = true
 re_format.workspace = true

crates/store/re_redap_client/src/connection_registry.rs

@@ -373,21 +373,24 @@ impl ConnectionRegistryHandle {
         // so if what we're trying to connect to is not a valid Rerun server, then cut out
         // a layer of noise:
         {
-            let res = match ehttp::fetch_async(ehttp::Request::get(format!(
-                "{}/version",
-                origin.as_url()
-            )))
-            .await
-            {
-                Ok(res) => res,
-                Err(err) => {
-                    let mut msg = format!("failed to connect to server '{origin}': {err}");
-                    if let Some(suggested) = suggest_api_prefix(&origin) {
-                        msg.push_str(&format!(". Did you mean '{suggested}'?"));
+            let res = crate::with_retry("http_version_fetch", || async {
+                match ehttp::fetch_async(ehttp::Request::get(format!(
+                    "{}/version",
+                    origin.as_url()
+                )))
+                .await
+                {
+                    Ok(res) => Ok(res),
+                    Err(err) => {
+                        let mut msg = format!("failed to connect to server '{origin}': {err}");
+                        if let Some(suggested) = suggest_api_prefix(&origin) {
+                            msg.push_str(&format!(". Did you mean '{suggested}'?"));
+                        }
+                        Err(ApiError::connection(msg))
                     }
-                    return Err(ApiError::connection(msg));
                 }
-            };
+            })
+            .await?;
 
             if !res.ok {
                 let hint = suggest_api_prefix(&origin).map(|suggested| {

crates/store/re_redap_client/src/grpc.rs

@@ -130,7 +130,10 @@ pub(crate) async fn client(
     origin: Origin,
     credentials: Option<Arc<dyn re_auth::credentials::CredentialsProvider + Send + Sync + 'static>>,
 ) -> ApiResult<RedapClient> {
-    let channel = channel(origin).await?;
+    let channel = crate::with_retry("redap_connection", || async {
+        channel(origin.clone()).await
+    })
+    .await?;
 
     let middlewares = tower::ServiceBuilder::new()
         .layer(AuthDecorator::new(credentials))
@@ -182,7 +185,10 @@ pub(crate) async fn client(
     origin: Origin,
     credentials: Option<Arc<dyn re_auth::credentials::CredentialsProvider + Send + Sync + 'static>>,
 ) -> ApiResult<RedapClient> {
-    let channel = channel(origin).await?;
+    let channel = crate::with_retry("redap_connection", || async {
+        channel(origin.clone()).await
+    })
+    .await?;
 
     let middlewares = tower::ServiceBuilder::new()
         .layer(AuthDecorator::new(credentials))

crates/store/re_redap_client/src/lib.rs

@@ -104,6 +104,7 @@ pub enum ApiErrorKind {
     Timeout,
     Internal,
     InvalidArguments,
+    ResourcesExhausted,
     Serialization,
     InvalidServer,
 }
@@ -114,6 +115,7 @@ impl From<tonic::Code> for ApiErrorKind {
             tonic::Code::NotFound => Self::NotFound,
             tonic::Code::AlreadyExists => Self::AlreadyExists,
             tonic::Code::PermissionDenied => Self::PermissionDenied,
+            tonic::Code::ResourceExhausted => Self::ResourcesExhausted,
             tonic::Code::Unauthenticated => Self::Unauthenticated,
             tonic::Code::Unimplemented => Self::Unimplemented,
             tonic::Code::Unavailable => Self::Connection,
@@ -135,6 +137,7 @@ impl std::fmt::Display for ApiErrorKind {
             Self::Connection => write!(f, "Connection"),
             Self::Internal => write!(f, "Internal"),
             Self::InvalidArguments => write!(f, "InvalidArguments"),
+            Self::ResourcesExhausted => write!(f, "ResourcesExhausted"),
             Self::Serialization => write!(f, "Serialization"),
             Self::Timeout => write!(f, "Timeout"),
             Self::InvalidServer => write!(f, "InvalidServer"),
@@ -282,3 +285,74 @@ impl std::error::Error for ApiError {
             .map(|e| e as &(dyn std::error::Error + 'static))
     }
 }
+
+/// Helper function for executing requests or connection attempts with retries.
+#[tracing::instrument(skip(f), level = "debug")]
+pub async fn with_retry<T, F, Fut>(req_name: &str, f: F) -> ApiResult<T>
+where
+    F: Fn() -> Fut,
+    Fut: Future<Output = ApiResult<T>>,
+{
+    // targeting to have all retries finish under ~5 seconds
+    const MAX_ATTEMPTS: usize = 5;
+
+    // 100 200 400 800 1600
+    let mut backoff_gen = re_backoff::BackoffGenerator::new(
+        std::time::Duration::from_millis(100),
+        std::time::Duration::from_secs(3),
+    )
+    .expect("base is less than max");
+
+    let mut attempts = 1;
+    let mut last_retryable_err = None;
+
+    while attempts <= MAX_ATTEMPTS {
+        let res = f().await;
+
+        match res {
+            Err(err)
+                if matches!(
+                    err.kind,
+                    ApiErrorKind::Connection
+                        | ApiErrorKind::Timeout
+                        | ApiErrorKind::Internal
+                        | ApiErrorKind::ResourcesExhausted
+                ) =>
+            {
+                last_retryable_err = Some(err);
+                let backoff = backoff_gen.gen_next();
+
+                tracing::debug!(
+                    attempts,
+                    max_attempts = MAX_ATTEMPTS,
+                    ?backoff,
+                    "{req_name} failed with retryable gRPC error, retrying after backoff"
+                );
+
+                backoff.sleep().await;
+            }
+            Err(err) => {
+                tracing::error!(
+                    attempts,
+                    "{req_name} failed with non-retryable error: {err}"
+                );
+                return Err(err);
+            }
+
+            Ok(value) => {
+                tracing::debug!(attempts, "{req_name} succeeded");
+                return Ok(value);
+            }
+        }
+
+        attempts += 1;
+    }
+
+    tracing::error!(
+        attempts,
+        max_attempts = MAX_ATTEMPTS,
+        "{req_name} failed after max retries, giving up"
+    );
+
+    Err(last_retryable_err.expect("bug: this should not be None if we reach here"))
+}

crates/utils/re_backoff/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "re_backoff"
+authors.workspace = true
+description = "Utilities for retrying operations with backoff."
+edition.workspace = true
+homepage.workspace = true
+include.workspace = true
+license.workspace = true
+publish = true
+readme = "README.md"
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[lints]
+workspace = true
+
+[package.metadata.docs.rs]
+all-features = true
+
+
+[dependencies]
+# External
+rand.workspace = true
+tokio = { workspace = true, features = ["time"] }
+
+
+[dev-dependencies]

crates/utils/re_backoff/README.md

@@ -0,0 +1,14 @@
+# re_backoff
+
+Part of the [`rerun`](https://github.com/rerun-io/rerun) family of crates.
+
+[![Latest version](https://img.shields.io/crates/v/re_arrow_combinators.svg)](https://crates.io/crates/re_arrow_combinators)
+[![Documentation](https://docs.rs/re_arrow_combinators/badge.svg)](https://docs.rs/re_arrow_combinators)
+![MIT](https://img.shields.io/badge/license-MIT-blue.svg)
+![Apache](https://img.shields.io/badge/license-Apache-blue.svg)
+
+Implements utility code to help with backoff and retry logic.
+
+### Why not use existing traits like backoff, backon, tokio-retry2, or tower(retry)?
+
+The code is small and simple, that it feels unnecessary to add an external dependency for it. We should re-evaluate should this become ever-complicated.