fix(daemon): use per-lane timeouts

crates/arr-daemon/src/reconcile.rs

@@ -20,7 +20,11 @@ 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);
+
+const RECONCILE_ACTION_TIMEOUT: Duration = Duration::from_secs(25);
+const REAPER_ACTION_TIMEOUT: Duration = Duration::from_mins(1);
+const RSS_ACTION_TIMEOUT: Duration = Duration::from_mins(5);
+const METADATA_ACTION_TIMEOUT: Duration = Duration::from_mins(30);
 
 pub type ActionError = Box<dyn Error + Send + Sync>;
 pub type ActionFuture<'a> =
@@ -117,6 +121,15 @@ impl Schedule {
             Tick::Reaper => REAPER_START_DELAY,
         }
     }
+
+    const fn action_timeout(tick: Tick) -> Duration {
+        match tick {
+            Tick::Reconcile => RECONCILE_ACTION_TIMEOUT,
+            Tick::Rss => RSS_ACTION_TIMEOUT,
+            Tick::Metadata => METADATA_ACTION_TIMEOUT,
+            Tick::Reaper => REAPER_ACTION_TIMEOUT,
+        }
+    }
 }
 
 /// The daemon's four-lane scheduler. Actions register into a lane; no queued
@@ -125,7 +138,7 @@ impl Schedule {
 pub struct ReconcileLoop {
     database: Db,
     schedule: Schedule,
-    action_timeout: Duration,
+    action_timeout_override: Option<Duration>,
     actions: Vec<RegisteredAction>,
 }
 
@@ -135,7 +148,7 @@ impl ReconcileLoop {
         Self {
             database,
             schedule: Schedule::default(),
-            action_timeout: ACTION_TIMEOUT,
+            action_timeout_override: None,
             actions: Vec::new(),
         }
     }
@@ -187,7 +200,15 @@ impl ReconcileLoop {
                     if *shutdown.borrow() {
                         return;
                     }
-                    self.run_tick(tick).await;
+                    tokio::select! {
+                        biased;
+                        changed = shutdown.changed() => {
+                            if changed.is_err() || *shutdown.borrow() {
+                                return;
+                            }
+                        }
+                        _ = self.run_tick(tick) => {}
+                    }
                 }
             }
         }
@@ -195,10 +216,12 @@ impl ReconcileLoop {
 
     async fn run_tick(&self, tick: Tick) -> TickReport {
         let mut report = TickReport::default();
+        let action_timeout = self
+            .action_timeout_override
+            .unwrap_or_else(|| Schedule::action_timeout(tick));
 
         for registered in self.actions.iter().filter(|action| action.tick == tick) {
-            match tokio::time::timeout(self.action_timeout, registered.action.run(&self.database))
-                .await
+            match tokio::time::timeout(action_timeout, registered.action.run(&self.database)).await
             {
                 Ok(Ok(outcomes)) => {
                     for outcome in outcomes {
@@ -226,7 +249,7 @@ impl ReconcileLoop {
                     tracing::error!(
                         tick = %tick,
                         worker = registered.action.name(),
-                        timeout_seconds = self.action_timeout.as_secs(),
+                        timeout_seconds = action_timeout.as_secs(),
                         "reconcile action timed out"
                     );
                 }
@@ -259,7 +282,7 @@ mod tests {
     use std::collections::HashMap;
     use std::io;
 
-    use tokio::sync::Mutex;
+    use tokio::sync::{Mutex, Notify};
 
     use super::*;
 
@@ -292,8 +315,10 @@ mod tests {
         fail_after_add: bool,
     }
 
-    #[derive(Debug)]
-    struct StalledAction;
+    #[derive(Clone, Debug, Default)]
+    struct StalledAction {
+        started: Arc<Notify>,
+    }
 
     impl Action for StalledAction {
         fn name(&self) -> &'static str {
@@ -301,7 +326,10 @@ mod tests {
         }
 
         fn run<'a>(&'a self, _database: &'a Db) -> ActionFuture<'a> {
-            Box::pin(std::future::pending())
+            Box::pin(async move {
+                self.started.notify_one();
+                std::future::pending().await
+            })
         }
     }
 
@@ -401,6 +429,20 @@ mod tests {
         );
         assert_eq!(Schedule::start_delay(Tick::Reaper), Duration::from_secs(10));
         assert_eq!(Schedule::start_delay(Tick::Rss), Duration::from_secs(20));
+
+        assert_eq!(
+            Schedule::action_timeout(Tick::Reconcile),
+            Duration::from_secs(25)
+        );
+        assert_eq!(
+            Schedule::action_timeout(Tick::Reaper),
+            Duration::from_mins(1)
+        );
+        assert_eq!(Schedule::action_timeout(Tick::Rss), Duration::from_mins(5));
+        assert_eq!(
+            Schedule::action_timeout(Tick::Metadata),
+            Duration::from_mins(30)
+        );
     }
 
     #[tokio::test]
@@ -445,12 +487,31 @@ mod tests {
     #[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 mut reconcile =
+            ReconcileLoop::new(database).register(Tick::Reconcile, StalledAction::default());
+        reconcile.action_timeout_override = Some(Duration::from_millis(1));
 
         let report = reconcile.run_tick(Tick::Reconcile).await;
 
         assert_eq!(report.failures, 1);
         assert_eq!(report.actions_taken, 0);
     }
+
+    #[tokio::test]
+    async fn shutdown_cancels_an_in_flight_action() {
+        let (_directory, database, _release_id) = seeded_database().await;
+        let action = StalledAction::default();
+        let started = Arc::clone(&action.started);
+        let reconcile = ReconcileLoop::new(database).register(Tick::Reconcile, action);
+        let (shutdown_tx, shutdown_rx) = watch::channel(false);
+        let task = tokio::spawn(reconcile.run(shutdown_rx));
+        started.notified().await;
+
+        shutdown_tx.send(true).unwrap();
+
+        tokio::time::timeout(Duration::from_millis(100), task)
+            .await
+            .expect("reconcile loop stopped")
+            .expect("reconcile task did not panic");
+    }
 }