fix(daemon): harden reconcile execution

crates/arr-daemon/src/main.rs

@@ -118,18 +118,29 @@ async fn run() -> Result<(), Error> {
     tracing::info!(addr = %config.bind_addr, docs = arr_api::DOCS_PATH, "listening");
 
     let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
-    let reconcile_task = tokio::spawn(reconcile.run(shutdown_rx));
+    let mut reconcile_task = tokio::spawn(reconcile.run(shutdown_rx));
     let signal_tx = shutdown_tx.clone();
-    let result = axum::serve(listener, app)
-        .with_graceful_shutdown(async move {
-            shutdown().await;
-            let _ = signal_tx.send(true);
-        })
-        .await;
+    let server = async move {
+        axum::serve(listener, app)
+            .with_graceful_shutdown(async move {
+                shutdown().await;
+                let _ = signal_tx.send(true);
+            })
+            .await
+    };
+    tokio::pin!(server);
 
-    let _ = shutdown_tx.send(true);
-    reconcile_task.await?;
-    result.map_err(Error::Serve)
+    tokio::select! {
+        result = &mut server => {
+            let _ = shutdown_tx.send(true);
+            reconcile_task.await?;
+            result.map_err(Error::Serve)
+        }
+        result = &mut reconcile_task => {
+            result?;
+            server.await.map_err(Error::Serve)
+        }
+    }
 }
 
 /// Stop accepting on Ctrl-C, or on the SIGTERM a service manager sends.

crates/arr-daemon/src/reconcile.rs

@@ -20,6 +20,7 @@ const RECONCILE_START_DELAY: Duration = Duration::ZERO;
 const METADATA_START_DELAY: Duration = Duration::from_secs(5);
 const REAPER_START_DELAY: Duration = Duration::from_secs(10);
 const RSS_START_DELAY: Duration = Duration::from_secs(20);
+const ACTION_TIMEOUT: Duration = Duration::from_secs(25);
 
 pub type ActionError = Box<dyn Error + Send + Sync>;
 pub type ActionFuture<'a> =
@@ -124,6 +125,7 @@ impl Schedule {
 pub struct ReconcileLoop {
     database: Db,
     schedule: Schedule,
+    action_timeout: Duration,
     actions: Vec<RegisteredAction>,
 }
 
@@ -133,6 +135,7 @@ impl ReconcileLoop {
         Self {
             database,
             schedule: Schedule::default(),
+            action_timeout: ACTION_TIMEOUT,
             actions: Vec::new(),
         }
     }
@@ -194,8 +197,10 @@ impl ReconcileLoop {
         let mut report = TickReport::default();
 
         for registered in self.actions.iter().filter(|action| action.tick == tick) {
-            match registered.action.run(&self.database).await {
-                Ok(outcomes) => {
+            match tokio::time::timeout(self.action_timeout, registered.action.run(&self.database))
+                .await
+            {
+                Ok(Ok(outcomes)) => {
                     for outcome in outcomes {
                         tracing::info!(
                             tick = %tick,
@@ -205,10 +210,9 @@ impl ReconcileLoop {
                             "reconcile action"
                         );
                         report.actions_taken += 1;
-                        report.gaps_found += 1;
                     }
                 }
-                Err(error) => {
+                Ok(Err(error)) => {
                     report.failures += 1;
                     tracing::error!(
                         tick = %tick,
@@ -217,23 +221,34 @@ impl ReconcileLoop {
                         "reconcile action failed"
                     );
                 }
+                Err(_) => {
+                    report.failures += 1;
+                    tracing::error!(
+                        tick = %tick,
+                        worker = registered.action.name(),
+                        timeout_seconds = self.action_timeout.as_secs(),
+                        "reconcile action timed out"
+                    );
+                }
             }
         }
 
-        tracing::info!(
-            tick = %tick,
-            gaps_found = report.gaps_found,
-            actions_taken = report.actions_taken,
-            failures = report.failures,
-            "reconcile tick"
-        );
+        if report.actions_taken == 0 && report.failures == 0 {
+            tracing::debug!(tick = %tick, "reconcile tick idle");
+        } else {
+            tracing::info!(
+                tick = %tick,
+                actions_taken = report.actions_taken,
+                failures = report.failures,
+                "reconcile tick"
+            );
+        }
         report
     }
 }
 
 #[derive(Debug, Default, PartialEq, Eq)]
 struct TickReport {
-    gaps_found: usize,
     actions_taken: usize,
     failures: usize,
 }
@@ -277,6 +292,19 @@ mod tests {
         fail_after_add: bool,
     }
 
+    #[derive(Debug)]
+    struct StalledAction;
+
+    impl Action for StalledAction {
+        fn name(&self) -> &'static str {
+            "stalled"
+        }
+
+        fn run<'a>(&'a self, _database: &'a Db) -> ActionFuture<'a> {
+            Box::pin(std::future::pending())
+        }
+    }
+
     impl Action for FakeGrabAction {
         fn name(&self) -> &'static str {
             "fake-grab"
@@ -413,4 +441,16 @@ mod tests {
             .unwrap();
         assert_eq!(grabs, 1);
     }
+
+    #[tokio::test]
+    async fn a_stalled_action_times_out() {
+        let (_directory, database, _release_id) = seeded_database().await;
+        let mut reconcile = ReconcileLoop::new(database).register(Tick::Reconcile, StalledAction);
+        reconcile.action_timeout = Duration::from_millis(1);
+
+        let report = reconcile.run_tick(Tick::Reconcile).await;
+
+        assert_eq!(report.failures, 1);
+        assert_eq!(report.actions_taken, 0);
+    }
 }