refactor(rust): make the auto-retry an implementation detail of repos…

…itories

examples/rust/get_started/src/token.rs

@@ -30,7 +30,7 @@ pub async fn create_token(attribute_name: &str, attribute_value: &str) -> Result
         ExportedEnrollmentTicket::from_str(token_string).map_err(|e| error(format!("could not decode token: {e}")))?;
     let ticket = decoded.import().await?;
 
-    Ok(ticket.one_time_code.clone())
+    Ok(ticket.one_time_code)
 }
 
 fn error(message: String) -> Error {

implementations/rust/ockam/ockam_abac/src/policy/storage/resource_policy_repository.rs

@@ -3,6 +3,10 @@ use ockam_core::async_trait;
 use ockam_core::compat::boxed::Box;
 use ockam_core::compat::vec::Vec;
 use ockam_core::Result;
+#[cfg(feature = "std")]
+use ockam_node::database::AutoRetry;
+#[cfg(feature = "std")]
+use ockam_node::retry;
 
 /// This repository stores policies for resources.
 /// A policy is an expression which can be evaluated against an environment (a list of attribute
@@ -36,3 +40,39 @@ pub trait ResourcePoliciesRepository: Send + Sync + 'static {
     /// Delete the policy associated to a given resource name and action
     async fn delete_policy(&self, resource_name: &ResourceName, action: &Action) -> Result<()>;
 }
+
+#[cfg(feature = "std")]
+#[async_trait]
+impl<T: ResourcePoliciesRepository> ResourcePoliciesRepository for AutoRetry<T> {
+    async fn store_policy(
+        &self,
+        resource_name: &ResourceName,
+        action: &Action,
+        expression: &Expr,
+    ) -> Result<()> {
+        retry!(self.wrapped.store_policy(resource_name, action, expression))
+    }
+
+    async fn get_policy(
+        &self,
+        resource_name: &ResourceName,
+        action: &Action,
+    ) -> Result<Option<ResourcePolicy>> {
+        retry!(self.wrapped.get_policy(resource_name, action))
+    }
+
+    async fn get_policies(&self) -> Result<Vec<ResourcePolicy>> {
+        retry!(self.wrapped.get_policies())
+    }
+
+    async fn get_policies_by_resource_name(
+        &self,
+        resource_name: &ResourceName,
+    ) -> Result<Vec<ResourcePolicy>> {
+        retry!(self.wrapped.get_policies_by_resource_name(resource_name))
+    }
+
+    async fn delete_policy(&self, resource_name: &ResourceName, action: &Action) -> Result<()> {
+        retry!(self.wrapped.delete_policy(resource_name, action))
+    }
+}

implementations/rust/ockam/ockam_abac/src/policy/storage/resource_policy_repository_sql.rs

@@ -1,14 +1,15 @@
 use core::str::FromStr;
 use sqlx::*;
+use std::sync::Arc;
 use tracing::debug;
 
+use crate::{Action, Expr, ResourceName, ResourcePoliciesRepository, ResourcePolicy};
 use ockam_core::async_trait;
 use ockam_core::compat::vec::Vec;
 use ockam_core::Result;
+use ockam_node::database::AutoRetry;
 use ockam_node::database::{FromSqlxError, SqlxDatabase, ToVoid};
 
-use crate::{Action, Expr, ResourceName, ResourcePoliciesRepository, ResourcePolicy};
-
 #[derive(Clone)]
 pub struct ResourcePolicySqlxDatabase {
     database: SqlxDatabase,
@@ -25,6 +26,18 @@ impl ResourcePolicySqlxDatabase {
         }
     }
 
+    /// Create a repository
+    pub fn make_repository(
+        database: SqlxDatabase,
+        node_name: &str,
+    ) -> Arc<dyn ResourcePoliciesRepository> {
+        if database.needs_retry() {
+            Arc::new(AutoRetry::new(Self::new(database, node_name)))
+        } else {
+            Arc::new(Self::new(database, node_name))
+        }
+    }
+
     /// Create a new in-memory database for policies
     pub async fn create() -> Result<Self> {
         Ok(Self::new(

implementations/rust/ockam/ockam_abac/src/policy/storage/resource_repository.rs

@@ -2,6 +2,10 @@ use crate::{Resource, ResourceName};
 use ockam_core::async_trait;
 use ockam_core::compat::boxed::Box;
 use ockam_core::Result;
+#[cfg(feature = "std")]
+use ockam_node::database::AutoRetry;
+#[cfg(feature = "std")]
+use ockam_node::retry;
 
 /// This repository stores resources.
 #[async_trait]
@@ -15,3 +19,19 @@ pub trait ResourcesRepository: Send + Sync + 'static {
     /// Delete all the entries for the given resource name
     async fn delete_resource(&self, resource_name: &ResourceName) -> Result<()>;
 }
+
+#[cfg(feature = "std")]
+#[async_trait]
+impl<T: ResourcesRepository> ResourcesRepository for AutoRetry<T> {
+    async fn store_resource(&self, resource: &Resource) -> Result<()> {
+        retry!(self.wrapped.store_resource(resource))
+    }
+
+    async fn get_resource(&self, resource_name: &ResourceName) -> Result<Option<Resource>> {
+        retry!(self.wrapped.get_resource(resource_name))
+    }
+
+    async fn delete_resource(&self, resource_name: &ResourceName) -> Result<()> {
+        retry!(self.wrapped.delete_resource(resource_name))
+    }
+}

implementations/rust/ockam/ockam_abac/src/policy/storage/resource_repository_sql.rs

@@ -3,14 +3,15 @@ use sqlx::encode::IsNull;
 use sqlx::error::BoxDynError;
 use sqlx::postgres::any::AnyArgumentBuffer;
 use sqlx::*;
+use std::sync::Arc;
 use tracing::debug;
 
+use crate::{Resource, ResourceName, ResourceType, ResourcesRepository};
 use ockam_core::async_trait;
 use ockam_core::Result;
+use ockam_node::database::AutoRetry;
 use ockam_node::database::{FromSqlxError, SqlxDatabase, ToVoid};
 
-use crate::{Resource, ResourceName, ResourceType, ResourcesRepository};
-
 #[derive(Clone)]
 pub struct ResourcesSqlxDatabase {
     database: SqlxDatabase,
@@ -27,6 +28,18 @@ impl ResourcesSqlxDatabase {
         }
     }
 
+    /// Create a repository
+    pub fn make_repository(
+        database: SqlxDatabase,
+        node_name: &str,
+    ) -> Arc<dyn ResourcesRepository> {
+        if database.needs_retry() {
+            Arc::new(AutoRetry::new(Self::new(database, node_name)))
+        } else {
+            Arc::new(Self::new(database, node_name))
+        }
+    }
+
     /// Create a new in-memory database for resources
     pub async fn create() -> Result<Self> {
         Ok(Self::new(

implementations/rust/ockam/ockam_abac/src/policy/storage/resource_type_policy_repository.rs

@@ -4,6 +4,10 @@ use ockam_core::async_trait;
 use ockam_core::compat::boxed::Box;
 use ockam_core::compat::vec::Vec;
 use ockam_core::Result;
+#[cfg(feature = "std")]
+use ockam_node::database::AutoRetry;
+#[cfg(feature = "std")]
+use ockam_node::retry;
 
 /// This repository stores policies for resources types.
 /// A policy is an expression which can be evaluated against an environment (a list of attribute
@@ -37,3 +41,39 @@ pub trait ResourceTypePoliciesRepository: Send + Sync + 'static {
     /// Delete the policy associated to a given resource type and action
     async fn delete_policy(&self, resource_type: &ResourceType, action: &Action) -> Result<()>;
 }
+
+#[cfg(feature = "std")]
+#[async_trait]
+impl<T: ResourceTypePoliciesRepository> ResourceTypePoliciesRepository for AutoRetry<T> {
+    async fn store_policy(
+        &self,
+        resource_type: &ResourceType,
+        action: &Action,
+        expression: &Expr,
+    ) -> Result<()> {
+        retry!(self.wrapped.store_policy(resource_type, action, expression))
+    }
+
+    async fn get_policy(
+        &self,
+        resource_type: &ResourceType,
+        action: &Action,
+    ) -> Result<Option<ResourceTypePolicy>> {
+        retry!(self.wrapped.get_policy(resource_type, action))
+    }
+
+    async fn get_policies(&self) -> Result<Vec<ResourceTypePolicy>> {
+        retry!(self.wrapped.get_policies())
+    }
+
+    async fn get_policies_by_resource_type(
+        &self,
+        resource_type: &ResourceType,
+    ) -> Result<Vec<ResourceTypePolicy>> {
+        retry!(self.wrapped.get_policies_by_resource_type(resource_type))
+    }
+
+    async fn delete_policy(&self, resource_type: &ResourceType, action: &Action) -> Result<()> {
+        retry!(self.wrapped.delete_policy(resource_type, action))
+    }
+}

implementations/rust/ockam/ockam_abac/src/policy/storage/resource_type_policy_repository_sql.rs

@@ -3,16 +3,17 @@ use sqlx::encode::IsNull;
 use sqlx::error::BoxDynError;
 use sqlx::postgres::any::AnyArgumentBuffer;
 use sqlx::*;
+use std::sync::Arc;
 use tracing::debug;
 
+use crate::policy::ResourceTypePolicy;
+use crate::{Action, Expr, ResourceType, ResourceTypePoliciesRepository};
 use ockam_core::async_trait;
 use ockam_core::compat::vec::Vec;
 use ockam_core::Result;
+use ockam_node::database::AutoRetry;
 use ockam_node::database::{FromSqlxError, SqlxDatabase, ToVoid};
 
-use crate::policy::ResourceTypePolicy;
-use crate::{Action, Expr, ResourceType, ResourceTypePoliciesRepository};
-
 #[derive(Clone)]
 pub struct ResourceTypePolicySqlxDatabase {
     database: SqlxDatabase,
@@ -29,6 +30,18 @@ impl ResourceTypePolicySqlxDatabase {
         }
     }
 
+    /// Create a repository
+    pub fn make_repository(
+        database: SqlxDatabase,
+        node_name: &str,
+    ) -> Arc<dyn ResourceTypePoliciesRepository> {
+        if database.needs_retry() {
+            Arc::new(AutoRetry::new(Self::new(database, node_name)))
+        } else {
+            Arc::new(Self::new(database, node_name))
+        }
+    }
+
     /// Create a new in-memory database for policies
     pub async fn create() -> Result<Self> {
         Ok(Self::new(

implementations/rust/ockam/ockam_api/src/authenticator/enrollment_tokens/issuer.rs

@@ -98,7 +98,7 @@ impl EnrollmentTokenIssuer {
         let now = now()?;
         let expires_at = now + max_token_duration.as_secs();
         let tkn = EnrollmentToken {
-            one_time_code: one_time_code.clone(),
+            one_time_code,
             reference: Some(reference.clone()),
             issued_by: enroller.clone(),
             created_at: now,

implementations/rust/ockam/ockam_api/src/authenticator/storage/authority_enrollment_token_repository.rs

@@ -4,6 +4,8 @@ use ockam::identity::TimestampInSeconds;
 use ockam_core::async_trait;
 use ockam_core::compat::boxed::Box;
 use ockam_core::Result;
+use ockam_node::database::AutoRetry;
+use ockam_node::retry;
 
 /// This repository stores enrollment tokens on the Authority node
 #[async_trait]
@@ -18,3 +20,18 @@ pub trait AuthorityEnrollmentTokenRepository: Send + Sync + 'static {
     /// Store a newly issued enrolment token
     async fn store_new_token(&self, token: EnrollmentToken) -> Result<()>;
 }
+
+#[async_trait]
+impl<T: AuthorityEnrollmentTokenRepository> AuthorityEnrollmentTokenRepository for AutoRetry<T> {
+    async fn use_token(
+        &self,
+        one_time_code: OneTimeCode,
+        now: TimestampInSeconds,
+    ) -> Result<Option<EnrollmentToken>> {
+        retry!(self.wrapped.use_token(one_time_code, now))
+    }
+
+    async fn store_new_token(&self, token: EnrollmentToken) -> Result<()> {
+        retry!(self.wrapped.store_new_token(token.clone()))
+    }
+}