Pr feedback - more rubust error handling

Author: Jamesbarford

database/src/pool.rs

@@ -190,14 +190,14 @@ pub trait Connection: Send + Sync {
         &self,
         statuses: &[BenchmarkRequestStatus],
         days: Option<i32>,
-    ) -> Vec<BenchmarkRequest>;
+    ) -> anyhow::Result<Vec<BenchmarkRequest>>;
 
     /// Update the status of a `benchmark_request`
     async fn update_benchmark_request_status(
         &mut self,
         benchmark_request: &BenchmarkRequest,
         benchmark_request_status: BenchmarkRequestStatus,
-    );
+    ) -> anyhow::Result<()>;
 }
 
 #[async_trait::async_trait]

database/src/pool/postgres.rs

@@ -1419,7 +1419,7 @@ where
         &self,
         statuses: &[BenchmarkRequestStatus],
         days: Option<i32>,
-    ) -> Vec<BenchmarkRequest> {
+    ) -> anyhow::Result<Vec<BenchmarkRequest>> {
         let mut query = "
                 SELECT
                     tag,
@@ -1441,15 +1441,14 @@ where
             // `::INT`, we can't do; `INTERVAL '$2 day'` which, while looking
             // natural is invalid.
             query += "AND created_at > current_date - $2::INT * INTERVAL '1 day'";
-            self.conn()
-                .query(&query, &[&statuses, &days])
-                .await
-                .unwrap()
+            self.conn().query(&query, &[&statuses, &days]).await
         } else {
-            self.conn().query(&query, &[&statuses]).await.unwrap()
-        };
+            self.conn().query(&query, &[&statuses]).await
+        }
+        .context("Failed to get benchmark requests")?;
 
-        rows.iter()
+        let benchmark_requests = rows
+            .iter()
             .map(|row| {
                 let tag = row.get::<_, String>(0);
                 let parent_sha = row.get::<_, Option<String>>(1);
@@ -1501,22 +1500,31 @@ where
                     ),
                 }
             })
-            .collect()
+            .collect();
+        Ok(benchmark_requests)
     }
 
     async fn update_benchmark_request_status(
         &mut self,
         benchmark_request: &BenchmarkRequest,
         benchmark_request_status: BenchmarkRequestStatus,
-    ) {
-        let tx = self.conn_mut().transaction().await.unwrap();
+    ) -> anyhow::Result<()> {
+        let tx = self
+            .conn_mut()
+            .transaction()
+            .await
+            .context("failed to start transaction")?;
+
         tx.execute(
             "UPDATE benchmark_request SET status = $1 WHERE tag = $2;",
             &[&benchmark_request_status, &benchmark_request.tag()],
         )
         .await
-        .unwrap();
-        tx.commit().await.unwrap();
+        .context("failed to execute UPDATE benchmark_request")?;
+
+        tx.commit().await.context("failed to commit transaction")?;
+
+        Ok(())
     }
 }
 

database/src/pool/sqlite.rs

@@ -1272,15 +1272,15 @@ impl Connection for SqliteConnection {
         &self,
         _statuses: &[BenchmarkRequestStatus],
         _days: Option<i32>,
-    ) -> Vec<BenchmarkRequest> {
+    ) -> anyhow::Result<Vec<BenchmarkRequest>> {
         no_queue_implementation_abort!()
     }
 
     async fn update_benchmark_request_status(
         &mut self,
         _benchmark_request: &BenchmarkRequest,
         _benchmark_request_status: BenchmarkRequestStatus,
-    ) {
+    ) -> anyhow::Result<()> {
         no_queue_implementation_abort!()
     }
 }

site/src/job_queue.rs

@@ -9,7 +9,7 @@ use tokio::time::{self, Duration};
 
 /// Store the latest master commits or do nothing if all of them are
 /// already in the database
-async fn enqueue_master_commits(ctxt: &Arc<SiteCtxt>) {
+async fn enqueue_master_commits(ctxt: &Arc<SiteCtxt>) -> anyhow::Result<()> {
     let conn = ctxt.conn().await;
     let master_commits = &ctxt.get_master_commits().commits;
     // TODO; delete at some point in the future
@@ -32,6 +32,7 @@ async fn enqueue_master_commits(ctxt: &Arc<SiteCtxt>) {
             conn.insert_benchmark_request(&benchmark).await;
         }
     }
+    Ok(())
 }
 
 // This function is split for testing purposes as mocking out `SiteCtxt` is non
@@ -86,7 +87,7 @@ fn get_next_benchmark_request(
 }
 
 /// Enqueue the job into the job_queue
-async fn enqueue_next_job(site_ctxt: &Arc<SiteCtxt>) {
+async fn enqueue_next_job(site_ctxt: &Arc<SiteCtxt>) -> anyhow::Result<()> {
     let mut conn = site_ctxt.conn().await;
     let mut pending = conn
         .get_benchmark_requests_by_status(
@@ -96,35 +97,38 @@ async fn enqueue_next_job(site_ctxt: &Arc<SiteCtxt>) {
             ],
             None,
         )
-        .await;
+        .await?;
 
     // No requests to process or we have something currently in progress
     if pending
         .iter()
         .any(|r| r.status == BenchmarkRequestStatus::InProgress)
     {
-        return;
+        return Ok(());
     }
 
     // We draw back the last 30 days of completed requests
     let completed = conn
         .get_benchmark_requests_by_status(&[BenchmarkRequestStatus::Completed], Some(30))
-        .await;
+        .await?;
 
     // And we now see if we have another request that can be processed
     if let Some(next_request) = get_next_benchmark_request(&mut pending, &completed) {
         // TODO; we simply flip the status for now however this should also
         // create the relevant jobs in the `job_queue`
         conn.update_benchmark_request_status(&next_request, BenchmarkRequestStatus::InProgress)
-            .await;
+            .await?
     }
+
+    Ok(())
 }
 
 /// For queueing jobs, add the jobs you want to queue to this function
-async fn cron_enqueue_jobs(site_ctxt: &Arc<SiteCtxt>) {
+async fn cron_enqueue_jobs(site_ctxt: &Arc<SiteCtxt>) -> anyhow::Result<()> {
     // Put the master commits into the `benchmark_requests` queue
-    enqueue_master_commits(site_ctxt).await;
-    enqueue_next_job(site_ctxt).await;
+    enqueue_master_commits(site_ctxt).await?;
+    enqueue_next_job(site_ctxt).await?;
+    Ok(())
 }
 
 /// Entry point for the cron
@@ -139,8 +143,10 @@ pub async fn cron_main(site_ctxt: Arc<RwLock<Option<Arc<SiteCtxt>>>>, seconds: u
             let guard = ctxt.read();
             guard.as_ref().cloned()
         } {
-            cron_enqueue_jobs(&ctxt_clone).await;
-            log::info!("Cron job executed at: {:?}", std::time::SystemTime::now());
+            match cron_enqueue_jobs(&ctxt_clone).await {
+                Ok(_) => log::info!("Cron job executed at: {:?}", std::time::SystemTime::now()),
+                Err(e) => log::error!("Cron job failed to execute {}", e),
+            }
         }
     }
 }