feat: retry dynamodb HttpDispatch errors (#605)

and add back retry logic to fetch message calls (legacy autopush and the
python version previously did this)

Note: the majority of HttpDispatch errors are likely due to a
misconfigured pool (so retrying is a lazy fix, but sufficient):

Closes: SYNC-4126

autopush-common/src/db/dynamodb/mod.rs

@@ -8,7 +8,7 @@ use crate::db::client::DbClient;
 use crate::db::dynamodb::retry::{
     retry_policy, retryable_batchwriteitem_error, retryable_delete_error,
     retryable_describe_table_error, retryable_getitem_error, retryable_putitem_error,
-    retryable_updateitem_error,
+    retryable_query_error, retryable_updateitem_error,
 };
 use crate::db::error::{DbError, DbResult};
 use crate::db::{
@@ -399,7 +399,13 @@ impl DbClient for DdbClientImpl {
             ..Default::default()
         };
 
-        let output = self.db_client.query(input.clone()).await?;
+        let output = retry_policy()
+            .retry_if(
+                || self.db_client.query(input.clone()),
+                retryable_query_error(self.metrics.clone()),
+            )
+            .await?;
+
         let mut notifs: Vec<NotificationRecord> = output.items.map_or_else(Vec::new, |items| {
             debug!("Got response of: {:?}", items);
             items
@@ -462,7 +468,13 @@ impl DbClient for DdbClientImpl {
             ..Default::default()
         };
 
-        let output = self.db_client.query(input.clone()).await?;
+        let output = retry_policy()
+            .retry_if(
+                || self.db_client.query(input.clone()),
+                retryable_query_error(self.metrics.clone()),
+            )
+            .await?;
+
         let messages = output.items.map_or_else(Vec::new, |items| {
             debug!("Got response of: {:?}", items);
             items

autopush-common/src/db/dynamodb/retry.rs

@@ -3,7 +3,7 @@ use cadence::{CountedExt, StatsdClient};
 use rusoto_core::RusotoError;
 use rusoto_dynamodb::{
     BatchWriteItemError, DeleteItemError, DescribeTableError, GetItemError, PutItemError,
-    UpdateItemError,
+    QueryError, UpdateItemError,
 };
 use std::sync::Arc;
 
@@ -12,7 +12,8 @@ macro_rules! retryable_error {
     ($name:ident, $error:tt, $error_tag:expr) => {
         pub fn $name(metrics: Arc<StatsdClient>) -> impl Fn(&RusotoError<$error>) -> bool {
             move |err| match err {
-                RusotoError::Service($error::InternalServerError(_))
+                RusotoError::HttpDispatch(_)
+                | RusotoError::Service($error::InternalServerError(_))
                 | RusotoError::Service($error::ProvisionedThroughputExceeded(_)) => {
                     debug!("retryable {} {:?}", $error_tag, &err);
                     metrics
@@ -27,6 +28,7 @@ macro_rules! retryable_error {
     };
 }
 
+retryable_error!(retryable_query_error, QueryError, "query");
 retryable_error!(retryable_getitem_error, GetItemError, "get_item");
 retryable_error!(retryable_updateitem_error, UpdateItemError, "update_item");
 retryable_error!(retryable_putitem_error, PutItemError, "put_item");