Properly handle errors in post indexer and profile classifier

Reconnect to Bluesky in the indexer
Don't exit the classifier just because we couldn't fetch profiles

README.md

@@ -16,7 +16,7 @@ Heavily WIP. Doesn't work yet at all, but does read the stream of posts as they
 - [x] Handle errors in the web service gracefully
 - [x] Handle missing profiles in the profile classifier
 - [x] Add a way to mark a profile as being from a certain country manually
-- [ ] Handle reconnecting to websocket somehow
+- [x] Handle reconnecting to websocket somehow
 - [ ] Publish the feed
 
 ## Configuration

src/processes/post_indexer.rs

@@ -1,8 +1,9 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::Result;
 use async_trait::async_trait;
-use log::info;
+use log::{error, info};
 
 use crate::algos::Algos;
 use crate::config::Config;
@@ -36,6 +37,18 @@ impl PostIndexer {
     pub async fn start(&self) -> Result<()> {
         info!("Starting");
 
+        loop {
+            if let Err(e) = self.process_from_last_point().await {
+                error!("Stopped because of an error: {}", e);
+            }
+
+            info!("Waiting 10 seconds before reconnecting...");
+
+            tokio::time::sleep(Duration::from_secs(10)).await;
+        }
+    }
+
+    async fn process_from_last_point(&self) -> Result<()> {
         let cursor = self
             .database
             .fetch_subscription_cursor(&self.config.feed_generator_did)

src/processes/profile_classifier.rs

@@ -25,9 +25,13 @@ impl ProfileClassifier {
 
     pub async fn start(&self) -> Result<()> {
         info!("Starting");
+
         loop {
-            // TODO: Don't just exit this function when an error happens, just wait a minute or so?
-            self.classify_unclassified_profiles().await?;
+            if let Err(e) = self.classify_unclassified_profiles().await {
+                error!("Problem with classifying profiles: {}", e)
+            }
+
+            tokio::time::sleep(Duration::from_secs(10)).await;
         }
     }
 
@@ -35,8 +39,9 @@ impl ProfileClassifier {
         // TODO: Maybe streamify this so that each thing is processed in parallel
 
         let dids = self.database.fetch_unprocessed_profile_dids().await?;
+
         if dids.is_empty() {
-            info!("No profiles to process: waiting 10 seconds");
+            info!("No profiles to process");
         } else {
             info!("Classifying {} new profiles", dids.len());
             for did in &dids {
@@ -47,8 +52,6 @@ impl ProfileClassifier {
             }
         }
 
-        tokio::time::sleep(Duration::from_secs(10)).await;
-
         Ok(())
     }