feat(rust): enable auto-retry on all repositories

implementations/rust/ockam/ockam_api/src/authenticator/one_time_code.rs

@@ -13,7 +13,7 @@ use std::fmt::{Debug, Formatter};
 /// A one-time code can be used to enroll
 /// a node with some authenticated attributes
 /// It can be retrieved with a command like `ockam project ticket --attribute component=control`
-#[derive(Clone, Encode, Decode, CborLen, PartialEq, Eq)]
+#[derive(Clone, Encode, Decode, CborLen, PartialEq, Eq, Copy)]
 #[rustfmt::skip]
 #[cbor(map)]
 pub struct OneTimeCode {

implementations/rust/ockam/ockam_api/src/authority_node/authority.rs

@@ -24,6 +24,7 @@ use ockam_node::database::SqlxDatabase;
 use ockam_node::Context;
 
 use crate::authority_node::Configuration;
+use crate::cli_state::AutoRetry;
 use crate::echoer::Echoer;
 use crate::nodes::service::default_address::DefaultAddress;
 
@@ -73,9 +74,15 @@ impl Authority {
             SqlxDatabase::create(&configuration.database_configuration).await?
         };
 
-        let members = Arc::new(AuthorityMembersSqlxDatabase::new(database.clone()));
-        let tokens = Arc::new(AuthorityEnrollmentTokenSqlxDatabase::new(database.clone()));
-        let secure_channel_repository = Arc::new(SecureChannelSqlxDatabase::new(database.clone()));
+        let members = Arc::new(AutoRetry::new(AuthorityMembersSqlxDatabase::new(
+            database.clone(),
+        )));
+        let tokens = Arc::new(AutoRetry::new(AuthorityEnrollmentTokenSqlxDatabase::new(
+            database.clone(),
+        )));
+        let secure_channel_repository = Arc::new(AutoRetry::new(SecureChannelSqlxDatabase::new(
+            database.clone(),
+        )));
 
         Self::bootstrap_repository(members.clone(), configuration).await?;
 

implementations/rust/ockam/ockam_api/src/cli_state/identities_attributes.rs

@@ -1,3 +1,4 @@
+use crate::cli_state::AutoRetry;
 use crate::CliState;
 use ockam::identity::{
     IdentitiesAttributes, IdentityAttributesRepository, IdentityAttributesSqlxDatabase,
@@ -18,9 +19,9 @@ impl CliState {
         &self,
         node_name: &str,
     ) -> Arc<dyn IdentityAttributesRepository> {
-        Arc::new(IdentityAttributesSqlxDatabase::new(
+        Arc::new(AutoRetry::new(IdentityAttributesSqlxDatabase::new(
             self.database(),
             node_name,
-        ))
+        )))
     }
 }

implementations/rust/ockam/ockam_api/src/cli_state/policies.rs

@@ -1,15 +1,18 @@
-use crate::cli_state::CliState;
+use crate::cli_state::{AutoRetry, CliState};
 use ockam_abac::{Policies, ResourcePolicySqlxDatabase, ResourceTypePolicySqlxDatabase};
 use std::sync::Arc;
 
 impl CliState {
     pub fn policies(&self, node_name: &str) -> Policies {
         Policies::new(
-            Arc::new(ResourcePolicySqlxDatabase::new(self.database(), node_name)),
-            Arc::new(ResourceTypePolicySqlxDatabase::new(
+            Arc::new(AutoRetry::new(ResourcePolicySqlxDatabase::new(
                 self.database(),
                 node_name,
-            )),
+            ))),
+            Arc::new(AutoRetry::new(ResourceTypePolicySqlxDatabase::new(
+                self.database(),
+                node_name,
+            ))),
         )
     }
 }

implementations/rust/ockam/ockam_api/src/cli_state/resources.rs

@@ -1,12 +1,13 @@
+use crate::cli_state::AutoRetry;
 use crate::CliState;
 use ockam_abac::{Resources, ResourcesSqlxDatabase};
 use std::sync::Arc;
 
 impl CliState {
     pub fn resources(&self, node_name: &str) -> Resources {
-        Resources::new(Arc::new(ResourcesSqlxDatabase::new(
+        Resources::new(Arc::new(AutoRetry::new(ResourcesSqlxDatabase::new(
             self.database(),
             node_name,
-        )))
+        ))))
     }
 }

implementations/rust/ockam/ockam_api/src/cli_state/secure_channels.rs

@@ -2,8 +2,8 @@ use std::sync::Arc;
 
 use ockam::identity::{Identities, SecureChannelSqlxDatabase, SecureChannels};
 
-use crate::cli_state::CliState;
 use crate::cli_state::Result;
+use crate::cli_state::{AutoRetry, CliState};
 
 impl CliState {
     pub async fn secure_channels(&self, node_name: &str) -> Result<Arc<SecureChannels>> {
@@ -15,7 +15,9 @@ impl CliState {
             .build();
         Ok(SecureChannels::from_identities(
             identities,
-            Arc::new(SecureChannelSqlxDatabase::new(self.database())),
+            Arc::new(AutoRetry::new(SecureChannelSqlxDatabase::new(
+                self.database(),
+            ))),
         ))
     }
 }

implementations/rust/ockam/ockam_api/src/cli_state/storage/auto_retry.rs

@@ -1,3 +1,8 @@
+use crate::authenticator::one_time_code::OneTimeCode;
+use crate::authenticator::{
+    AuthorityEnrollmentTokenRepository, AuthorityMember, AuthorityMembersRepository,
+    EnrollmentToken, PreTrustedIdentities,
+};
 use crate::cli_state::journeys::{Journey, ProjectJourney};
 use crate::cli_state::{
     EnrollmentsRepository, IdentitiesRepository, IdentityEnrollment, JourneysRepository,
@@ -14,9 +19,14 @@ use chrono::{DateTime, Utc};
 use ockam::identity::models::{ChangeHistory, CredentialAndPurposeKey, PurposeKeyAttestation};
 use ockam::identity::storage::PurposeKeysRepository;
 use ockam::identity::{
-    ChangeHistoryRepository, CredentialRepository, Identifier, Identity, Purpose,
+    AttributesEntry, ChangeHistoryRepository, CredentialRepository, Identifier, Identity,
+    IdentityAttributesRepository, PersistedSecureChannel, Purpose, SecureChannelRepository,
     TimestampInSeconds,
 };
+use ockam_abac::{
+    Action, Expr, Resource, ResourceName, ResourcePoliciesRepository, ResourcePolicy, ResourceType,
+    ResourceTypePoliciesRepository, ResourceTypePolicy, ResourcesRepository,
+};
 use ockam_core::{async_trait, Address};
 use ockam_vault::storage::SecretsRepository;
 use ockam_vault::{
@@ -44,7 +54,7 @@ macro_rules! retry {
 }
 
 #[derive(Clone)]
-pub(crate) struct AutoRetry<T: Sized + Send + Sync + 'static> {
+pub struct AutoRetry<T: Sized + Send + Sync + 'static> {
     wrapped: T,
 }
 
@@ -83,6 +93,21 @@ impl<T: EnrollmentsRepository> EnrollmentsRepository for AutoRetry<T> {
     }
 }
 
+#[async_trait]
+impl<T: AuthorityEnrollmentTokenRepository> AuthorityEnrollmentTokenRepository for AutoRetry<T> {
+    async fn use_token(
+        &self,
+        one_time_code: OneTimeCode,
+        now: TimestampInSeconds,
+    ) -> ockam_core::Result<Option<EnrollmentToken>> {
+        retry!(self.wrapped.use_token(one_time_code, now))
+    }
+
+    async fn store_new_token(&self, token: EnrollmentToken) -> ockam_core::Result<()> {
+        retry!(self.wrapped.store_new_token(token.clone()))
+    }
+}
+
 #[async_trait]
 impl<T: IdentitiesRepository + Send + Sync + 'static> IdentitiesRepository for AutoRetry<T> {
     async fn store_named_identity(
@@ -626,3 +651,171 @@ impl<T: SecretsRepository> SecretsRepository for AutoRetry<T> {
         retry!(self.wrapped.delete_all())
     }
 }
+
+#[async_trait]
+impl<T: ResourcesRepository> ResourcesRepository for AutoRetry<T> {
+    async fn store_resource(&self, resource: &Resource) -> ockam_core::Result<()> {
+        retry!(self.wrapped.store_resource(resource))
+    }
+
+    async fn get_resource(
+        &self,
+        resource_name: &ResourceName,
+    ) -> ockam_core::Result<Option<Resource>> {
+        retry!(self.wrapped.get_resource(resource_name))
+    }
+
+    async fn delete_resource(&self, resource_name: &ResourceName) -> ockam_core::Result<()> {
+        retry!(self.wrapped.delete_resource(resource_name))
+    }
+}
+
+#[async_trait]
+impl<T: ResourcePoliciesRepository> ResourcePoliciesRepository for AutoRetry<T> {
+    async fn store_policy(
+        &self,
+        resource_name: &ResourceName,
+        action: &Action,
+        expression: &Expr,
+    ) -> ockam_core::Result<()> {
+        retry!(self.wrapped.store_policy(resource_name, action, expression))
+    }
+
+    async fn get_policy(
+        &self,
+        resource_name: &ResourceName,
+        action: &Action,
+    ) -> ockam_core::Result<Option<ResourcePolicy>> {
+        retry!(self.wrapped.get_policy(resource_name, action))
+    }
+
+    async fn get_policies(&self) -> ockam_core::Result<Vec<ResourcePolicy>> {
+        retry!(self.wrapped.get_policies())
+    }
+
+    async fn get_policies_by_resource_name(
+        &self,
+        resource_name: &ResourceName,
+    ) -> ockam_core::Result<Vec<ResourcePolicy>> {
+        retry!(self.wrapped.get_policies_by_resource_name(resource_name))
+    }
+
+    async fn delete_policy(
+        &self,
+        resource_name: &ResourceName,
+        action: &Action,
+    ) -> ockam_core::Result<()> {
+        retry!(self.wrapped.delete_policy(resource_name, action))
+    }
+}
+
+#[async_trait]
+impl<T: ResourceTypePoliciesRepository> ResourceTypePoliciesRepository for AutoRetry<T> {
+    async fn store_policy(
+        &self,
+        resource_type: &ResourceType,
+        action: &Action,
+        expression: &Expr,
+    ) -> ockam_core::Result<()> {
+        retry!(self.wrapped.store_policy(resource_type, action, expression))
+    }
+
+    async fn get_policy(
+        &self,
+        resource_type: &ResourceType,
+        action: &Action,
+    ) -> ockam_core::Result<Option<ResourceTypePolicy>> {
+        retry!(self.wrapped.get_policy(resource_type, action))
+    }
+
+    async fn get_policies(&self) -> ockam_core::Result<Vec<ResourceTypePolicy>> {
+        retry!(self.wrapped.get_policies())
+    }
+
+    async fn get_policies_by_resource_type(
+        &self,
+        resource_type: &ResourceType,
+    ) -> ockam_core::Result<Vec<ResourceTypePolicy>> {
+        retry!(self.wrapped.get_policies_by_resource_type(resource_type))
+    }
+
+    async fn delete_policy(
+        &self,
+        resource_type: &ResourceType,
+        action: &Action,
+    ) -> ockam_core::Result<()> {
+        retry!(self.wrapped.delete_policy(resource_type, action))
+    }
+}
+
+#[async_trait]
+impl<T: AuthorityMembersRepository> AuthorityMembersRepository for AutoRetry<T> {
+    async fn get_member(
+        &self,
+        identifier: &Identifier,
+    ) -> ockam_core::Result<Option<AuthorityMember>> {
+        retry!(self.wrapped.get_member(identifier))
+    }
+
+    async fn get_members(&self) -> ockam_core::Result<Vec<AuthorityMember>> {
+        retry!(self.wrapped.get_members())
+    }
+
+    async fn delete_member(&self, identifier: &Identifier) -> ockam_core::Result<()> {
+        retry!(self.wrapped.delete_member(identifier))
+    }
+
+    async fn add_member(&self, member: AuthorityMember) -> ockam_core::Result<()> {
+        retry!(self.wrapped.add_member(member.clone()))
+    }
+
+    async fn bootstrap_pre_trusted_members(
+        &self,
+        pre_trusted_identities: &PreTrustedIdentities,
+    ) -> ockam_core::Result<()> {
+        retry!(self
+            .wrapped
+            .bootstrap_pre_trusted_members(pre_trusted_identities))
+    }
+}
+
+#[async_trait]
+impl<T: IdentityAttributesRepository> IdentityAttributesRepository for AutoRetry<T> {
+    async fn get_attributes(
+        &self,
+        subject: &Identifier,
+        attested_by: &Identifier,
+    ) -> ockam_core::Result<Option<AttributesEntry>> {
+        retry!(self.wrapped.get_attributes(subject, attested_by))
+    }
+
+    async fn put_attributes(
+        &self,
+        subject: &Identifier,
+        entry: AttributesEntry,
+    ) -> ockam_core::Result<()> {
+        retry!(self.wrapped.put_attributes(subject, entry.clone()))
+    }
+
+    async fn delete_expired_attributes(&self, now: TimestampInSeconds) -> ockam_core::Result<()> {
+        retry!(self.wrapped.delete_expired_attributes(now))
+    }
+}
+
+#[async_trait]
+impl<T: SecureChannelRepository> SecureChannelRepository for AutoRetry<T> {
+    async fn get(
+        &self,
+        decryptor_remote_address: &Address,
+    ) -> ockam_core::Result<Option<PersistedSecureChannel>> {
+        retry!(self.wrapped.get(decryptor_remote_address))
+    }
+
+    async fn put(&self, secure_channel: PersistedSecureChannel) -> ockam_core::Result<()> {
+        retry!(self.wrapped.put(secure_channel.clone()))
+    }
+
+    async fn delete(&self, decryptor_remote_address: &Address) -> ockam_core::Result<()> {
+        retry!(self.wrapped.delete(decryptor_remote_address))
+    }
+}

implementations/rust/ockam/ockam_api/src/cli_state/storage/mod.rs

@@ -18,7 +18,7 @@ pub use users_repository_sql::*;
 pub use vaults_repository::*;
 pub use vaults_repository_sql::*;
 
-mod auto_retry;
+pub mod auto_retry;
 mod enrollments_repository;
 mod enrollments_repository_sql;
 mod identities_repository;