chore: enable to retry in all operations. (#278)

Author: mythrnr

.github/workflows/ci.yml

@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@master
-      - uses: denoland/setup-deno@909cc5acb0fdd60627fb858598759246509fa755 # v2.0.2
+      - uses: denoland/setup-deno@e95548e56dfa95d4e1a28d6f422fafe75c4c26fb # v2.0.3
         with:
           deno-version: v1.x
       - name: Run deno fmt and deno lint

Cargo.lock

@@ -16,7 +16,7 @@ convert_case = "^0.8.0"
 ident_case = "^1.0.1"
 proc-macro2 = "^1.0.95"
 quote = "^1.0.40"
-syn = "^2.0.101"
+syn = "^2.0.104"
 
 [features]
 default = []

raiden-derive/Cargo.toml

@@ -48,6 +48,8 @@ pub(crate) fn expand_batch_delete(
                         client: &self.client,
                         write_requests,
                         table_name: self.table_name(),
+                        policy: self.retry_condition.strategy.policy(),
+                        condition: &self.retry_condition,
                     }
                 }
             }
@@ -86,6 +88,8 @@ pub(crate) fn expand_batch_delete(
                         client: &self.client,
                         write_requests,
                         table_name: self.table_name(),
+                        policy: self.retry_condition.strategy.policy(),
+                        condition: &self.retry_condition,
                     }
                 }
             }
@@ -95,12 +99,12 @@ pub(crate) fn expand_batch_delete(
     let api_call_token = super::api_call_token!("batch_write_item");
     let (call_inner_run, inner_run_args) = if cfg!(feature = "tracing") {
         (
-            quote! { #builder_name::inner_run(&self.table_name, &self.client, input).await? },
-            quote! { table_name: &str, },
+            quote! { #builder_name::inner_run(table_name, client, input).await },
+            quote! { table_name: String, },
         )
     } else {
         (
-            quote! { #builder_name::inner_run(&self.client, input).await? },
+            quote! { #builder_name::inner_run(client, input).await },
             quote! {},
         )
     };
@@ -112,17 +116,22 @@ pub(crate) fn expand_batch_delete(
             pub client: &'a ::raiden::DynamoDbClient,
             pub write_requests: std::vec::Vec<::raiden::WriteRequest>,
             pub table_name: String,
+            pub policy: ::raiden::Policy,
+            pub condition: &'a ::raiden::retry::RetryCondition,
         }
 
         impl<'a> #builder_name<'a> {
-            pub async fn run(mut self) -> Result<::raiden::batch_delete::BatchDeleteOutput, ::raiden::RaidenError> {
+            pub async fn run(self) -> Result<::raiden::batch_delete::BatchDeleteOutput, ::raiden::RaidenError> {
+                let Self { client, mut write_requests, table_name, policy, condition } = self;
+                let policy: ::raiden::RetryPolicy = policy.into();
+
                 // TODO: set the number of retry to 5 for now, which should be made more flexible
                 const RETRY: usize = 5;
                 const MAX_ITEMS_PER_REQUEST: usize = 25;
 
                 for _ in 0..RETRY {
                     loop {
-                        let len = self.write_requests.len();
+                        let len = write_requests.len();
 
                         // len == 0 means there are no items to be processed anymore
                         if len == 0 {
@@ -132,16 +141,24 @@ pub(crate) fn expand_batch_delete(
                         let start = len.saturating_sub(MAX_ITEMS_PER_REQUEST);
                         let end = std::cmp::min(len, start + MAX_ITEMS_PER_REQUEST);
                         // take requests up to 25 from the request buffer
-                        let req = self.write_requests.drain(start..end).collect::<std::vec::Vec<_>>();
-                        let request_items = vec![(self.table_name.clone(), req)]
+                        let req = write_requests.drain(start..end).collect::<std::vec::Vec<_>>();
+                        let request_items = vec![(table_name.clone(), req)]
                             .into_iter()
                             .collect::<std::collections::HashMap<_, _>>();
-                        let input = ::raiden::BatchWriteItemInput {
-                            request_items,
-                            ..std::default::Default::default()
-                        };
+                        let result = {
+                            let t = table_name.clone();
+                            let c = client.clone();
+                            let i = ::raiden::BatchWriteItemInput {
+                                request_items,
+                                ..std::default::Default::default()
+                            };
 
-                        let result = #call_inner_run;
+                            policy.retry_if(move || {
+                                let (table_name, client, input)
+                                    = (t.clone(), c.clone(), i.clone());
+                                async move { #call_inner_run }
+                            }, condition).await?
+                        };
 
                         let mut unprocessed_items = match result.unprocessed_items {
                             None => {
@@ -161,16 +178,16 @@ pub(crate) fn expand_batch_delete(
                         };
 
                         let unprocessed_requests = unprocessed_items
-                            .remove(&self.table_name)
-                            .expect("reqeust_items hashmap must have a value for the table name");
+                            .remove(&table_name)
+                            .expect("request_items hashmap must have a value for the table name");
                         // push unprocessed requests back to the request buffer
-                        self.write_requests.extend(unprocessed_requests);
+                        write_requests.extend(unprocessed_requests);
                     }
                 }
 
                 // when retry is done the specified times, treat it as success even if there are
                 // still unprocessed items
-                let unprocessed_items = self.write_requests
+                let unprocessed_items = write_requests
                     .into_iter()
                     .filter_map(|write_request| write_request.delete_request)
                     .collect::<std::vec::Vec<_>>();
@@ -182,7 +199,7 @@ pub(crate) fn expand_batch_delete(
 
             async fn inner_run(
                 #inner_run_args
-                client: &::raiden::DynamoDbClient,
+                client: ::raiden::DynamoDbClient,
                 input: ::raiden::BatchWriteItemInput,
             ) -> Result<::raiden::BatchWriteItemOutput, ::raiden::RaidenError> {
                 Ok(#api_call_token?)