diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 2758bd5..4299ccc 100644 --- a/crates/arr-daemon/src/main.rs +++ b/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. diff --git a/crates/arr-daemon/src/reconcile.rs b/crates/arr-daemon/src/reconcile.rs index d849533..7a742b0 100644 --- a/crates/arr-daemon/src/reconcile.rs +++ b/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; pub type ActionFuture<'a> = @@ -124,6 +125,7 @@ impl Schedule { pub struct ReconcileLoop { database: Db, schedule: Schedule, + action_timeout: Duration, actions: Vec, } @@ -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); + } }