From 79516eedeceb4fbdacba7b2aa230e6766799a1ae Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 21:26:51 +0100 Subject: [PATCH 1/4] feat(daemon): add reconcile scheduler --- crates/arr-daemon/src/main.rs | 22 +- crates/arr-daemon/src/reconcile.rs | 399 +++++++++++++++++++++++++++++ 2 files changed, 417 insertions(+), 4 deletions(-) create mode 100644 crates/arr-daemon/src/reconcile.rs diff --git a/crates/arr-daemon/src/main.rs b/crates/arr-daemon/src/main.rs index 6d9b587..cc1a571 100644 --- a/crates/arr-daemon/src/main.rs +++ b/crates/arr-daemon/src/main.rs @@ -1,6 +1,7 @@ //! arr — reconcile loop and process entry point. See DESIGN.md §8. mod config; +pub mod reconcile; mod web; use std::process::ExitCode; @@ -11,6 +12,7 @@ use arr_compat::CompatState; use arr_db::Db; use arr_meta::TmdbClient; use config::Config; +use reconcile::ReconcileLoop; use tower_http::trace::TraceLayer; /// Dump the `OpenAPI` document and exit, instead of serving. `just gen-client` @@ -72,12 +74,15 @@ enum Error { }, #[error("serve: {0}")] Serve(std::io::Error), + #[error("reconcile task: {0}")] + ReconcileTask(#[from] tokio::task::JoinError), } async fn run() -> Result<(), Error> { let config = Config::load()?; let database = Db::connect(&config.database_path).await?; database.migrate().await?; + let reconcile = ReconcileLoop::new(database.clone()); // Jellyseerr's Radarr shim (DESIGN.md §9.4) reads the same database and // needs its own TMDB client for `movie/lookup`. @@ -107,10 +112,19 @@ async fn run() -> Result<(), Error> { tracing::info!(addr = %config.bind_addr, docs = arr_api::DOCS_PATH, "listening"); - axum::serve(listener, app) - .with_graceful_shutdown(shutdown()) - .await - .map_err(Error::Serve) + let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); + let 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 _ = shutdown_tx.send(true); + reconcile_task.await?; + result.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 new file mode 100644 index 0000000..5dfb39d --- /dev/null +++ b/crates/arr-daemon/src/reconcile.rs @@ -0,0 +1,399 @@ +//! Periodic desired-state reconciliation. See DESIGN.md §8. + +use std::error::Error; +use std::fmt; +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; + +use arr_db::Db; +use tokio::sync::watch; +use tokio::time::{Instant, MissedTickBehavior}; + +pub const RECONCILE_INTERVAL: Duration = Duration::from_secs(30); +pub const RSS_INTERVAL: Duration = Duration::from_mins(10); +pub const METADATA_INTERVAL: Duration = Duration::from_hours(24); +pub const REAPER_INTERVAL: Duration = Duration::from_mins(5); + +pub type ActionError = Box; +pub type ActionFuture<'a> = + Pin, ActionError>> + Send + 'a>>; + +/// One independently scheduled class of reconciliation work. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Tick { + Reconcile, + Rss, + Metadata, + Reaper, +} + +impl fmt::Display for Tick { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(match self { + Self::Reconcile => "reconcile", + Self::Rss => "rss", + Self::Metadata => "metadata", + Self::Reaper => "reaper", + }) + } +} + +/// A desired/actual gap closed by an action during a tick. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Outcome { + gap: String, + action: String, +} + +impl Outcome { + #[must_use] + pub fn new(gap: impl Into, action: impl Into) -> Self { + Self { + gap: gap.into(), + action: action.into(), + } + } +} + +/// A concrete reconciler registered for one tick class. +/// +/// Implementations query persistent domain rows on every call. They may keep +/// in-flight state in memory, but the first call must rebuild it from the +/// external system so restarting the process converges without a job table. +pub trait Action: fmt::Debug + Send + Sync { + fn name(&self) -> &'static str; + + fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a>; +} + +#[derive(Debug)] +struct RegisteredAction { + tick: Tick, + action: Arc, +} + +#[derive(Debug, Clone, Copy)] +struct Schedule { + reconcile: Duration, + rss: Duration, + metadata: Duration, + reaper: Duration, +} + +impl Default for Schedule { + fn default() -> Self { + Self { + reconcile: RECONCILE_INTERVAL, + rss: RSS_INTERVAL, + metadata: METADATA_INTERVAL, + reaper: REAPER_INTERVAL, + } + } +} + +impl Schedule { + const fn interval(self, tick: Tick) -> Duration { + match tick { + Tick::Reconcile => self.reconcile, + Tick::Rss => self.rss, + Tick::Metadata => self.metadata, + Tick::Reaper => self.reaper, + } + } +} + +/// The daemon's four-lane scheduler. Actions register into a lane; no queued +/// work is persisted because each action rediscovers gaps from domain rows. +#[derive(Debug)] +pub struct ReconcileLoop { + database: Db, + schedule: Schedule, + actions: Vec, +} + +impl ReconcileLoop { + #[must_use] + pub fn new(database: Db) -> Self { + Self { + database, + schedule: Schedule::default(), + actions: Vec::new(), + } + } + + /// Register an action in the cadence that owns it. + #[must_use] + pub fn register(mut self, tick: Tick, action: A) -> Self + where + A: Action + 'static, + { + self.actions.push(RegisteredAction { + tick, + action: Arc::new(action), + }); + self + } + + /// Run until shutdown. Reconciliation runs immediately on startup so + /// actions can reconstruct transient state before serving later ticks. + pub async fn run(self, shutdown: watch::Receiver) { + let this = Arc::new(self); + let reconcile = Arc::clone(&this).run_lane(Tick::Reconcile, true, shutdown.clone()); + let rss = Arc::clone(&this).run_lane(Tick::Rss, false, shutdown.clone()); + let metadata = Arc::clone(&this).run_lane(Tick::Metadata, false, shutdown.clone()); + let reaper = Arc::clone(&this).run_lane(Tick::Reaper, false, shutdown); + + tokio::join!(reconcile, rss, metadata, reaper); + } + + async fn run_lane( + self: Arc, + tick: Tick, + run_immediately: bool, + mut shutdown: watch::Receiver, + ) { + if *shutdown.borrow() { + return; + } + + let interval = self.schedule.interval(tick); + let first_tick = if run_immediately { + Instant::now() + } else { + Instant::now() + interval + }; + let mut timer = tokio::time::interval_at(first_tick, interval); + timer.set_missed_tick_behavior(MissedTickBehavior::Skip); + + loop { + tokio::select! { + changed = shutdown.changed() => { + if changed.is_err() || *shutdown.borrow() { + return; + } + } + _ = timer.tick() => { + self.run_tick(tick).await; + } + } + } + } + + async fn run_tick(&self, tick: Tick) -> TickReport { + 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) => { + for outcome in outcomes { + tracing::info!( + tick = %tick, + worker = registered.action.name(), + gap = %outcome.gap, + action = %outcome.action, + "reconcile action" + ); + report.actions_taken += 1; + report.gaps_found += 1; + } + } + Err(error) => { + report.failures += 1; + tracing::error!( + tick = %tick, + worker = registered.action.name(), + error = %error, + "reconcile action failed" + ); + } + } + } + + tracing::info!( + tick = %tick, + gaps_found = report.gaps_found, + 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, +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use std::collections::HashMap; + use std::io; + + use tokio::sync::Mutex; + + use super::*; + + #[derive(Clone, Debug, Default)] + struct FakeTransmission { + torrents: Arc>>, + } + + impl FakeTransmission { + async fn find_or_add(&self, movie_id: i64) -> (String, bool) { + let mut torrents = self.torrents.lock().await; + if let Some(hash) = torrents.get(&movie_id) { + return (hash.clone(), false); + } + + let hash = format!("fake-{movie_id}"); + torrents.insert(movie_id, hash.clone()); + (hash, true) + } + + async fn len(&self) -> usize { + self.torrents.lock().await.len() + } + } + + #[derive(Debug)] + struct FakeGrabAction { + transmission: FakeTransmission, + release_id: i64, + fail_after_add: bool, + } + + impl Action for FakeGrabAction { + fn name(&self) -> &'static str { + "fake-grab" + } + + fn run<'a>(&'a self, database: &'a Db) -> ActionFuture<'a> { + Box::pin(async move { + let movie_id = sqlx::query_scalar::<_, i64>( + "SELECT id FROM movies m + WHERE wanted = 1 AND blocked = 0 AND state = 'missing' + AND NOT EXISTS ( + SELECT 1 FROM grabs g + WHERE g.target_kind = 'movie' AND g.target_id = m.id + AND g.state IN ('sent', 'downloaded', 'imported') + ) + ORDER BY id LIMIT 1", + ) + .fetch_optional(database.pool()) + .await?; + let Some(movie_id) = movie_id else { + return Ok(Vec::new()); + }; + + let (infohash, added) = self.transmission.find_or_add(movie_id).await; + if added && self.fail_after_add { + return Err(Box::new(io::Error::new( + io::ErrorKind::Interrupted, + "simulated process death after Transmission accepted the torrent", + )) as ActionError); + } + + sqlx::query( + "INSERT INTO grabs (release_id, target_kind, target_id, infohash) + VALUES (?, 'movie', ?, ?)", + ) + .bind(self.release_id) + .bind(movie_id) + .bind(infohash) + .execute(database.pool()) + .await?; + + Ok(vec![Outcome::new( + format!("wanted movie {movie_id} has no active grab"), + if added { + "added torrent and recorded grab" + } else { + "recovered existing torrent and recorded grab" + }, + )]) + }) + } + } + + async fn seeded_database() -> (tempfile::TempDir, Db, i64) { + let directory = tempfile::tempdir().unwrap(); + let database = Db::connect(directory.path().join("arr.db")).await.unwrap(); + database.migrate().await.unwrap(); + + let root_id: i64 = + sqlx::query_scalar("SELECT id FROM roots WHERE kind = 'movie' AND audience = 'main'") + .fetch_one(database.pool()) + .await + .unwrap(); + sqlx::query("INSERT INTO movies (tmdb_id, title, root_id) VALUES (1, 'Movie', ?)") + .bind(root_id) + .execute(database.pool()) + .await + .unwrap(); + let release_id = sqlx::query( + "INSERT INTO releases + (indexer_id, guid, name, size, download_url, parsed, verdict) + VALUES (1, 'guid', 'Movie.1080p', 1, 'magnet:test', '{}', 'eligible')", + ) + .execute(database.pool()) + .await + .unwrap() + .last_insert_rowid(); + + (directory, database, release_id) + } + + #[test] + fn default_schedule_matches_the_design() { + let schedule = Schedule::default(); + assert_eq!(schedule.interval(Tick::Reconcile), Duration::from_secs(30)); + assert_eq!(schedule.interval(Tick::Rss), Duration::from_mins(10)); + assert_eq!(schedule.interval(Tick::Metadata), Duration::from_hours(24)); + assert_eq!(schedule.interval(Tick::Reaper), Duration::from_mins(5)); + } + + #[tokio::test] + async fn restart_converges_without_duplicate_external_work() { + let (_directory, database, release_id) = seeded_database().await; + let transmission = FakeTransmission::default(); + + let interrupted = ReconcileLoop::new(database.clone()).register( + Tick::Reconcile, + FakeGrabAction { + transmission: transmission.clone(), + release_id, + fail_after_add: true, + }, + ); + let first = interrupted.run_tick(Tick::Reconcile).await; + assert_eq!(first.failures, 1); + assert_eq!(transmission.len().await, 1); + drop(interrupted); + + let restarted = ReconcileLoop::new(database.clone()).register( + Tick::Reconcile, + FakeGrabAction { + transmission: transmission.clone(), + release_id, + fail_after_add: false, + }, + ); + let recovered = restarted.run_tick(Tick::Reconcile).await; + let stable = restarted.run_tick(Tick::Reconcile).await; + + assert_eq!(recovered.actions_taken, 1); + assert_eq!(stable.actions_taken, 0); + assert_eq!(transmission.len().await, 1); + let grabs: i64 = sqlx::query_scalar("SELECT count(*) FROM grabs") + .fetch_one(database.pool()) + .await + .unwrap(); + assert_eq!(grabs, 1); + } +} -- 2.47.3 From d541ea0d985e66b24b9df226f8c098e1e98da513 Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 21:32:05 +0100 Subject: [PATCH 2/4] fix(daemon): stagger reconcile lanes --- crates/arr-daemon/src/reconcile.rs | 47 ++++++++++++++++++++---------- 1 file changed, 32 insertions(+), 15 deletions(-) diff --git a/crates/arr-daemon/src/reconcile.rs b/crates/arr-daemon/src/reconcile.rs index 5dfb39d..d849533 100644 --- a/crates/arr-daemon/src/reconcile.rs +++ b/crates/arr-daemon/src/reconcile.rs @@ -16,6 +16,11 @@ pub const RSS_INTERVAL: Duration = Duration::from_mins(10); pub const METADATA_INTERVAL: Duration = Duration::from_hours(24); pub const REAPER_INTERVAL: Duration = Duration::from_mins(5); +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); + pub type ActionError = Box; pub type ActionFuture<'a> = Pin, ActionError>> + Send + 'a>>; @@ -102,6 +107,15 @@ impl Schedule { Tick::Reaper => self.reaper, } } + + const fn start_delay(tick: Tick) -> Duration { + match tick { + Tick::Reconcile => RECONCILE_START_DELAY, + Tick::Rss => RSS_START_DELAY, + Tick::Metadata => METADATA_START_DELAY, + Tick::Reaper => REAPER_START_DELAY, + } + } } /// The daemon's four-lane scheduler. Actions register into a lane; no queued @@ -140,41 +154,36 @@ impl ReconcileLoop { /// actions can reconstruct transient state before serving later ticks. pub async fn run(self, shutdown: watch::Receiver) { let this = Arc::new(self); - let reconcile = Arc::clone(&this).run_lane(Tick::Reconcile, true, shutdown.clone()); - let rss = Arc::clone(&this).run_lane(Tick::Rss, false, shutdown.clone()); - let metadata = Arc::clone(&this).run_lane(Tick::Metadata, false, shutdown.clone()); - let reaper = Arc::clone(&this).run_lane(Tick::Reaper, false, shutdown); + let reconcile = Arc::clone(&this).run_lane(Tick::Reconcile, shutdown.clone()); + let rss = Arc::clone(&this).run_lane(Tick::Rss, shutdown.clone()); + let metadata = Arc::clone(&this).run_lane(Tick::Metadata, shutdown.clone()); + let reaper = Arc::clone(&this).run_lane(Tick::Reaper, shutdown); tokio::join!(reconcile, rss, metadata, reaper); } - async fn run_lane( - self: Arc, - tick: Tick, - run_immediately: bool, - mut shutdown: watch::Receiver, - ) { + async fn run_lane(self: Arc, tick: Tick, mut shutdown: watch::Receiver) { if *shutdown.borrow() { return; } let interval = self.schedule.interval(tick); - let first_tick = if run_immediately { - Instant::now() - } else { - Instant::now() + interval - }; + let first_tick = Instant::now() + Schedule::start_delay(tick); let mut timer = tokio::time::interval_at(first_tick, interval); timer.set_missed_tick_behavior(MissedTickBehavior::Skip); loop { tokio::select! { + biased; changed = shutdown.changed() => { if changed.is_err() || *shutdown.borrow() { return; } } _ = timer.tick() => { + if *shutdown.borrow() { + return; + } self.run_tick(tick).await; } } @@ -356,6 +365,14 @@ mod tests { assert_eq!(schedule.interval(Tick::Rss), Duration::from_mins(10)); assert_eq!(schedule.interval(Tick::Metadata), Duration::from_hours(24)); assert_eq!(schedule.interval(Tick::Reaper), Duration::from_mins(5)); + + assert_eq!(Schedule::start_delay(Tick::Reconcile), Duration::ZERO); + assert_eq!( + Schedule::start_delay(Tick::Metadata), + Duration::from_secs(5) + ); + assert_eq!(Schedule::start_delay(Tick::Reaper), Duration::from_secs(10)); + assert_eq!(Schedule::start_delay(Tick::Rss), Duration::from_secs(20)); } #[tokio::test] -- 2.47.3 From d8455df62d0100ffd7b7794ed8ba87753e37340f Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 21:41:14 +0100 Subject: [PATCH 3/4] fix(daemon): harden reconcile execution --- crates/arr-daemon/src/main.rs | 31 ++++++++++----- crates/arr-daemon/src/reconcile.rs | 64 ++++++++++++++++++++++++------ 2 files changed, 73 insertions(+), 22 deletions(-) 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); + } } -- 2.47.3 From 76e398d0b479c8c45f5617a49e0e8bd5ded6ce0b Mon Sep 17 00:00:00 2001 From: Miguel Palhas Date: Sat, 22 Aug 2026 21:44:59 +0100 Subject: [PATCH 4/4] 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"); + } } -- 2.47.3