Updates unmaintained tokio-retry to tokio-retry2

foundation/gax/Cargo.toml

@@ -17,4 +17,4 @@ thiserror = "1.0"
 tower = { version = "0.4", features = ["filter"] }
 http = "1.1"
 google-cloud-token = { version = "0.1.2", path = "../token" }
-tokio-retry = "0.3"
+tokio-retry2 = "0.5.3"

foundation/gax/src/retry.rs

@@ -2,9 +2,9 @@ use std::future::Future;
 use std::iter::Take;
 use std::time::Duration;
 
-pub use tokio_retry::strategy::ExponentialBackoff;
-pub use tokio_retry::Condition;
-use tokio_retry::{Action, RetryIf};
+pub use tokio_retry2::strategy::ExponentialBackoff;
+use tokio_retry2::{Action, RetryIf};
+pub use tokio_retry2::{Condition, MapErr};
 
 use crate::grpc::{Code, Status};
 
@@ -21,6 +21,7 @@ impl TryAs<Status> for Status {
 pub trait Retry<E: TryAs<Status>, T: Condition<E>> {
     fn strategy(&self) -> Take<ExponentialBackoff>;
     fn condition(&self) -> T;
+    fn notify(error: &E, duration: Duration);
 }
 
 pub struct CodeCondition {
@@ -70,6 +71,10 @@ impl Retry<Status, CodeCondition> for RetrySetting {
     fn condition(&self) -> CodeCondition {
         CodeCondition::new(self.codes.clone())
     }
+
+    fn notify(_error: &Status, _duration: Duration) {
+        tracing::trace!("retry fn");
+    }
 }
 
 impl Default for RetrySetting {
@@ -92,7 +97,7 @@ where
     RT: Retry<E, C> + Default,
 {
     let retry = retry.unwrap_or_default();
-    RetryIf::spawn(retry.strategy(), action, retry.condition()).await
+    RetryIf::spawn(retry.strategy(), action, retry.condition(), RT::notify).await
 }
 /// Repeats retries when the specified error is detected.
 /// The argument specified by 'v' can be reused for each retry.
@@ -117,7 +122,6 @@ where
         if retry.condition().should_retry(&status) {
             let duration = strategy.next().ok_or(status)?;
             tokio::time::sleep(duration).await;
-            tracing::trace!("retry fn");
         } else {
             return Err(status);
         }
@@ -128,6 +132,7 @@ where
 mod tests {
     use std::sync::{Arc, Mutex};
 
+    use tokio_retry2::MapErr;
     use tonic::{Code, Status};
 
     use crate::retry::{invoke, RetrySetting};
@@ -140,7 +145,7 @@ mod tests {
             let mut lock = counter.lock().unwrap();
             *lock += 1;
             let result: Result<i32, Status> = Err(Status::new(Code::Aborted, "error"));
-            result
+            result.map_transient_err()
         };
         let actual = invoke(Some(retry), action).await.unwrap_err();
         let expected = Status::new(Code::Aborted, "error");

foundation/longrunning/src/autogen/operations_client.rs

@@ -5,7 +5,7 @@ use tonic::Response;
 use google_cloud_gax::conn::{Channel, Error};
 use google_cloud_gax::create_request;
 use google_cloud_gax::grpc::{Code, Status};
-use google_cloud_gax::retry::{invoke, RetrySetting};
+use google_cloud_gax::retry::{invoke, MapErr, RetrySetting};
 use google_cloud_googleapis::longrunning::operations_client::OperationsClient as InternalOperationsClient;
 use google_cloud_googleapis::longrunning::{
     CancelOperationRequest, DeleteOperationRequest, GetOperationRequest, Operation, WaitOperationRequest,
@@ -44,7 +44,7 @@ impl OperationsClient {
         let name = &req.name;
         let action = || async {
             let request = create_request(format!("name={name}"), req.clone());
-            self.inner.clone().get_operation(request).await
+            self.inner.clone().get_operation(request).await.map_transient_err()
         };
         invoke(Some(setting), action).await
     }
@@ -62,7 +62,7 @@ impl OperationsClient {
         let name = &req.name;
         let action = || async {
             let request = create_request(format!("name={name}"), req.clone());
-            self.inner.clone().delete_operation(request).await
+            self.inner.clone().delete_operation(request).await.map_transient_err()
         };
         invoke(Some(setting), action).await
     }
@@ -86,7 +86,7 @@ impl OperationsClient {
         let name = &req.name;
         let action = || async {
             let request = create_request(format!("name={name}"), req.clone());
-            self.inner.clone().cancel_operation(request).await
+            self.inner.clone().cancel_operation(request).await.map_transient_err()
         };
         invoke(Some(setting), action).await
     }
@@ -108,7 +108,7 @@ impl OperationsClient {
         let setting = retry.unwrap_or_else(default_retry_setting);
         let action = || async {
             let request = create_request("".to_string(), req.clone());
-            self.inner.clone().wait_operation(request).await
+            self.inner.clone().wait_operation(request).await.map_transient_err()
         };
         invoke(Some(setting), action).await
     }

kms/src/grpc/apiv1/kms_client.rs

@@ -5,7 +5,7 @@ use crate::grpc::apiv1::conn_pool::ConnectionManager;
 
 use google_cloud_gax::create_request;
 use google_cloud_gax::grpc::{Code, Status};
-use google_cloud_gax::retry::{invoke, RetrySetting};
+use google_cloud_gax::retry::{invoke, MapErr, RetrySetting};
 
 use crate::grpc::kms::v1::CreateCryptoKeyVersionRequest;
 use crate::grpc::kms::v1::CreateKeyRingRequest;
@@ -62,7 +62,7 @@ impl Client {
     ) -> Result<GenerateRandomBytesResponse, Status> {
         let action = || async {
             let request = create_request(format!("location={}", req.location), req.clone());
-            self.cm.conn().generate_random_bytes(request).await
+            self.cm.conn().generate_random_bytes(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -81,7 +81,7 @@ impl Client {
     ) -> Result<CryptoKey, Status> {
         let action = || async {
             let request = create_request(format!("parent={}", req.parent), req.clone());
-            self.cm.conn().create_crypto_key(request).await
+            self.cm.conn().create_crypto_key(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -100,7 +100,11 @@ impl Client {
     ) -> Result<CryptoKeyVersion, Status> {
         let action = || async {
             let request = create_request(format!("parent={}", req.parent), req.clone());
-            self.cm.conn().create_crypto_key_version(request).await
+            self.cm
+                .conn()
+                .create_crypto_key_version(request)
+                .await
+                .map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -119,7 +123,7 @@ impl Client {
     ) -> Result<KeyRing, Status> {
         let action = || async {
             let request = create_request(format!("parent={}", req.parent), req.clone());
-            self.cm.conn().create_key_ring(request).await
+            self.cm.conn().create_key_ring(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -138,7 +142,11 @@ impl Client {
     ) -> Result<CryptoKeyVersion, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().destroy_crypto_key_version(request).await
+            self.cm
+                .conn()
+                .destroy_crypto_key_version(request)
+                .await
+                .map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -157,7 +165,7 @@ impl Client {
     ) -> Result<CryptoKey, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().get_crypto_key(request).await
+            self.cm.conn().get_crypto_key(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -176,7 +184,7 @@ impl Client {
     ) -> Result<CryptoKeyVersion, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().get_crypto_key_version(request).await
+            self.cm.conn().get_crypto_key_version(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -191,7 +199,7 @@ impl Client {
     pub async fn get_key_ring(&self, req: GetKeyRingRequest, retry: Option<RetrySetting>) -> Result<KeyRing, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().get_key_ring(request).await
+            self.cm.conn().get_key_ring(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -210,7 +218,11 @@ impl Client {
     ) -> Result<ListCryptoKeyVersionsResponse, Status> {
         let action = || async {
             let request = create_request(format!("parent={}", req.parent), req.clone());
-            self.cm.conn().list_crypto_key_versions(request).await
+            self.cm
+                .conn()
+                .list_crypto_key_versions(request)
+                .await
+                .map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -229,7 +241,7 @@ impl Client {
     ) -> Result<ListCryptoKeysResponse, Status> {
         let action = || async {
             let request = create_request(format!("parent={}", req.parent), req.clone());
-            self.cm.conn().list_crypto_keys(request).await
+            self.cm.conn().list_crypto_keys(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -248,7 +260,7 @@ impl Client {
     ) -> Result<ListKeyRingsResponse, Status> {
         let action = || async {
             let request = create_request(format!("parent={}", req.parent), req.clone());
-            self.cm.conn().list_key_rings(request).await
+            self.cm.conn().list_key_rings(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -263,7 +275,7 @@ impl Client {
     pub async fn encrypt(&self, req: EncryptRequest, retry: Option<RetrySetting>) -> Result<EncryptResponse, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().encrypt(request).await
+            self.cm.conn().encrypt(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -278,7 +290,7 @@ impl Client {
     pub async fn decrypt(&self, req: DecryptRequest, retry: Option<RetrySetting>) -> Result<DecryptResponse, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().decrypt(request).await
+            self.cm.conn().decrypt(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -297,7 +309,7 @@ impl Client {
     ) -> Result<AsymmetricSignResponse, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().asymmetric_sign(request).await
+            self.cm.conn().asymmetric_sign(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -312,7 +324,7 @@ impl Client {
     pub async fn mac_sign(&self, req: MacSignRequest, retry: Option<RetrySetting>) -> Result<MacSignResponse, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().mac_sign(request).await
+            self.cm.conn().mac_sign(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -331,7 +343,7 @@ impl Client {
     ) -> Result<MacVerifyResponse, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().mac_verify(request).await
+            self.cm.conn().mac_verify(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await
@@ -350,7 +362,7 @@ impl Client {
     ) -> Result<PublicKey, Status> {
         let action = || async {
             let request = create_request(format!("name={}", req.name), req.clone());
-            self.cm.conn().get_public_key(request).await
+            self.cm.conn().get_public_key(request).await.map_transient_err()
         };
         invoke(Some(retry.unwrap_or_else(default_setting)), action)
             .await

pubsub/src/apiv1/publisher_client.rs

@@ -4,7 +4,7 @@ use google_cloud_gax::conn::Channel;
 use google_cloud_gax::create_request;
 use google_cloud_gax::grpc::Response;
 use google_cloud_gax::grpc::{Code, Status};
-use google_cloud_gax::retry::{invoke, RetrySetting};
+use google_cloud_gax::retry::{invoke, MapErr, RetrySetting};
 use google_cloud_googleapis::pubsub::v1::publisher_client::PublisherClient as InternalPublisherClient;
 use google_cloud_googleapis::pubsub::v1::{
     DeleteTopicRequest, DetachSubscriptionRequest, DetachSubscriptionResponse, GetTopicRequest,
@@ -41,7 +41,7 @@ impl PublisherClient {
         let action = || async {
             let mut client = self.client();
             let request = create_request(format!("name={name}"), req.clone());
-            client.create_topic(request).await
+            client.create_topic(request).await.map_transient_err()
         };
         invoke(retry, action).await
     }
@@ -61,7 +61,7 @@ impl PublisherClient {
         let action = || async {
             let mut client = self.client();
             let request = create_request(format!("name={name}"), req.clone());
-            client.update_topic(request).await
+            client.update_topic(request).await.map_transient_err()
         };
         invoke(retry, action).await
     }
@@ -92,7 +92,7 @@ impl PublisherClient {
         let action = || async {
             let mut client = self.client();
             let request = create_request(format!("name={name}"), req.clone());
-            client.publish(request).await
+            client.publish(request).await.map_transient_err()
         };
         invoke(Some(setting), action).await
     }
@@ -108,7 +108,7 @@ impl PublisherClient {
         let action = || async {
             let mut client = self.client();
             let request = create_request(format!("topic={topic}"), req.clone());
-            client.get_topic(request).await
+            client.get_topic(request).await.map_transient_err()
         };
         invoke(retry, action).await
     }
@@ -127,7 +127,11 @@ impl PublisherClient {
             let action = || async {
                 let mut client = self.client();
                 let request = create_request(format!("project={project}"), req.clone());
-                client.list_topics(request).await.map(|d| d.into_inner())
+                client
+                    .list_topics(request)
+                    .await
+                    .map(|d| d.into_inner())
+                    .map_transient_err()
             };
             let response = invoke(retry.clone(), action).await?;
             all.extend(response.topics.into_iter());
@@ -152,7 +156,11 @@ impl PublisherClient {
             let action = || async {
                 let mut client = self.client();
                 let request = create_request(format!("topic={topic}"), req.clone());
-                client.list_topic_subscriptions(request).await.map(|d| d.into_inner())
+                client
+                    .list_topic_subscriptions(request)
+                    .await
+                    .map(|d| d.into_inner())
+                    .map_transient_err()
             };
             let response = invoke(retry.clone(), action).await?;
             all.extend(response.subscriptions.into_iter());
@@ -181,7 +189,11 @@ impl PublisherClient {
             let action = || async {
                 let mut client = self.client();
                 let request = create_request(format!("topic={topic}"), req.clone());
-                client.list_topic_snapshots(request).await.map(|d| d.into_inner())
+                client
+                    .list_topic_snapshots(request)
+                    .await
+                    .map(|d| d.into_inner())
+                    .map_transient_err()
             };
             let response = invoke(retry.clone(), action).await?;
             all.extend(response.snapshots.into_iter());
@@ -207,7 +219,7 @@ impl PublisherClient {
         let action = || async {
             let mut client = self.client();
             let request = create_request(format!("topic={topic}"), req.clone());
-            client.delete_topic(request).await
+            client.delete_topic(request).await.map_transient_err()
         };
         invoke(retry, action).await
     }
@@ -226,7 +238,7 @@ impl PublisherClient {
         let action = || async {
             let mut client = self.client();
             let request = create_request(format!("subscription={subscription}"), req.clone());
-            client.detach_subscription(request).await
+            client.detach_subscription(request).await.map_transient_err()
         };
         invoke(retry, action).await
     }