From 76e398d0b479c8c45f5617a49e0e8bd5ded6ce0b Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 21:44:59 +0100 Subject: [PATCH] fix(daemon): use per-lane timeouts --- crates/arr-daemon/src/reconcile.rs | 87 +++++++++++++++++++++++++----- 1 file changed, 74 insertions(+), 13 deletions(-) diff --git a/crates/arr-daemon/src/reconcile.rs b/crates/arr-daemon/src/reconcile.rs index 7a742b0..6fc7f15 100644 --- a/crates/arr-daemon/src/reconcile.rs +++ b/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; 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, actions: Vec, } @@ -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, + } 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"); + } }