Append final errors to error array and treat request errors as retryable

hook-common/src/pgqueue.rs

@@ -248,6 +248,7 @@ UPDATE
 SET
     finished_at = NOW(),
     status = 'failed'::job_status
+    errors = array_append("{0}".errors, $3)
 WHERE
     "{0}".id = $2
     AND queue = $1
@@ -261,6 +262,7 @@ RETURNING
         sqlx::query(&base_query)
             .bind(&failed_job.queue)
             .bind(failed_job.id)
+            .bind(&failed_job.error)
             .execute(&mut *self.connection)
             .await
             .map_err(|error| PgJobError::QueryError {
@@ -394,6 +396,7 @@ UPDATE
 SET
     finished_at = NOW(),
     status = 'failed'::job_status
+    errors = array_append("{0}".errors, $3)
 WHERE
     "{0}".id = $2
     AND queue = $1
@@ -406,6 +409,7 @@ RETURNING
         sqlx::query(&base_query)
             .bind(&failed_job.queue)
             .bind(failed_job.id)
+            .bind(&failed_job.error)
             .execute(&mut *self.transaction)
             .await
             .map_err(|error| PgJobError::QueryError {

hook-consumer/src/consumer.rs

@@ -32,7 +32,7 @@ impl<'p> WebhookConsumer<'p> {
         poll_interval: time::Duration,
         request_timeout: time::Duration,
         max_concurrent_jobs: usize,
-    ) -> Result<Self, WebhookConsumerError> {
+    ) -> Self {
         let mut headers = header::HeaderMap::new();
         headers.insert(
             header::CONTENT_TYPE,
@@ -42,15 +42,16 @@ impl<'p> WebhookConsumer<'p> {
         let client = reqwest::Client::builder()
             .default_headers(headers)
             .timeout(request_timeout)
-            .build()?;
+            .build()
+            .expect("failed to construct reqwest client for webhook consumer");
 
-        Ok(Self {
+        Self {
             name: name.to_owned(),
             queue,
             poll_interval,
             client,
             max_concurrent_jobs,
-        })
+        }
     }
 
     /// Wait until a job becomes available in our queue.
@@ -174,7 +175,11 @@ async fn send_webhook(
         .headers(headers)
         .body(body)
         .send()
-        .await?;
+        .await
+        .map_err(|e| WebhookConsumerError::RetryableWebhookError {
+            reason: e.to_string(),
+            retry_after: None,
+        })?;
 
     let status = response.status();
 
@@ -328,8 +333,8 @@ mod tests {
             time::Duration::from_millis(100),
             time::Duration::from_millis(5000),
             10,
-        )
-        .expect("consumer failed to initialize");
+        );
+
         let consumed_job = consumer
             .wait_for_job()
             .await

hook-consumer/src/error.rs

@@ -18,8 +18,6 @@ pub enum WebhookConsumerError {
     QueueError(#[from] pgqueue::PgQueueError),
     #[error("an error occurred in the underlying job")]
     PgJobError(String),
-    #[error("an error occurred when attempting to send a request")]
-    RequestError(#[from] reqwest::Error),
     #[error("a webhook could not be delivered but it could be retried later: {reason}")]
     RetryableWebhookError {
         reason: String,

hook-consumer/src/main.rs

@@ -29,7 +29,7 @@ async fn main() -> Result<(), WebhookConsumerError> {
         config.poll_interval.0,
         config.request_timeout.0,
         config.max_concurrent_jobs,
-    )?;
+    );
 
     let _ = consumer.run().await;