Survive transient errors in scanner and processor loops

A single failed GitHub request in the scanner loop (e.g. a transient
401 from an expired installation token) previously propagated out of
the task and terminated the entire rfd-processor process. The same
applied to a failed database query when fetching the job batch in the
processor loop.

Log these errors and retry on the next tick instead. Job creation is
idempotent (unique on sha + rfd) and the scanner re-enumerates all RFDs
every pass, so skipped iterations are recovered on the next successful
scan.

Author: notpeter

rfd-processor/src/processor.rs

@@ -39,14 +39,24 @@ pub async fn processor(ctx: Arc<Context>) -> Result<(), JobError> {
 
     loop {
         if ctx.processor.enabled {
-            let jobs = JobStore::list(
+            // Errors fetching the job list (i.e. transient database failures) must not take down
+            // the processor. Skip this batch and try again on the next tick
+            let jobs = match JobStore::list(
                 &ctx.db.storage,
                 vec![JobFilter::default()
                     .processed(Some(false))
                     .started(Some(false))],
                 &pagination,
             )
-            .await?;
+            .await
+            {
+                Ok(jobs) => jobs,
+                Err(err) => {
+                    tracing::error!(?err, "Failed to fetch job batch");
+                    interval.tick().await;
+                    continue;
+                }
+            };
 
             tracing::info!(jobs = ?jobs.iter().map(|job| job.id).collect::<Vec<_>>(), "Spawning jobs");
 

rfd-processor/src/scanner.rs

@@ -25,28 +25,36 @@ pub async fn scanner(ctx: Arc<Context>) -> Result<(), ScannerError> {
 
     loop {
         if ctx.scanner.enabled {
-            let updates = ctx
+            // Errors reaching GitHub (including transient auth and rate limit failures) must not
+            // take down the scanner. Skip this scan and try again on the next tick
+            match ctx
                 .github
                 .repository
                 .get_rfd_sync_updates(&ctx.github.client)
-                .await?;
-
-            for update in updates {
-                match JobStore::upsert(&ctx.db.storage, update.clone().into_job()).await {
-                    Ok(job) => tracing::trace!(?job.id, "Added job to the queue"),
-                    Err(err) => {
-                        match err {
-                            StoreError::Conflict => {
-                                // Nothing to do here, we expect uniqueness conflicts. It is expected
-                                // that the scanner picks ups redundant jobs for RFDs that have not
-                                // changed since the last scan
-                            }
-                            err => {
-                                tracing::warn!(?err, ?update, "Failed to add job")
+                .await
+            {
+                Ok(updates) => {
+                    for update in updates {
+                        match JobStore::upsert(&ctx.db.storage, update.clone().into_job()).await {
+                            Ok(job) => tracing::trace!(?job.id, "Added job to the queue"),
+                            Err(err) => {
+                                match err {
+                                    StoreError::Conflict => {
+                                        // Nothing to do here, we expect uniqueness conflicts. It is expected
+                                        // that the scanner picks ups redundant jobs for RFDs that have not
+                                        // changed since the last scan
+                                    }
+                                    err => {
+                                        tracing::warn!(?err, ?update, "Failed to add job")
+                                    }
+                                }
                             }
                         }
                     }
                 }
+                Err(err) => {
+                    tracing::error!(?err, "Failed to fetch RFD updates from GitHub");
+                }
             }
         }