From e5f4e449251510e0282f86b4df1ff36b6f8783fe Mon Sep 17 00:00:00 2001 From: "naps62-yolo (agent)" Date: Sat, 22 Aug 2026 22:00:12 +0100 Subject: [PATCH] Add reconcile loop scheduler (#69) --- crates/arr-daemon/src/main.rs | 33 +- crates/arr-daemon/src/reconcile.rs | 517 +++++++++++++++++++++++++++++ 2 files changed, 546 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 2004fc2..4299ccc 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`. @@ -112,10 +117,30 @@ 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 mut reconcile_task = tokio::spawn(reconcile.run(shutdown_rx)); + let signal_tx = shutdown_tx.clone(); + let server = async move { + axum::serve(listener, app) + .with_graceful_shutdown(async move { + shutdown().await; + let _ = signal_tx.send(true); + }) + .await + }; + tokio::pin!(server); + + 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 new file mode 100644 index 0000000..6fc7f15 --- /dev/null +++ b/crates/arr-daemon/src/reconcile.rs @@ -0,0 +1,517 @@ +//! 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); + +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 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> = + 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, + } + } + + 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, + } + } + + 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 +/// work is persisted because each action rediscovers gaps from domain rows. +#[derive(Debug)] +pub struct ReconcileLoop { + database: Db, + schedule: Schedule, + action_timeout_override: Option, + actions: Vec, +} + +impl ReconcileLoop { + #[must_use] + pub fn new(database: Db) -> Self { + Self { + database, + schedule: Schedule::default(), + action_timeout_override: None, + 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, 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, mut shutdown: watch::Receiver) { + if *shutdown.borrow() { + return; + } + + let interval = self.schedule.interval(tick); + 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; + } + tokio::select! { + biased; + changed = shutdown.changed() => { + if changed.is_err() || *shutdown.borrow() { + return; + } + } + _ = self.run_tick(tick) => {} + } + } + } + } + } + + 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(action_timeout, registered.action.run(&self.database)).await + { + Ok(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; + } + } + Ok(Err(error)) => { + report.failures += 1; + tracing::error!( + tick = %tick, + worker = registered.action.name(), + error = %error, + "reconcile action failed" + ); + } + Err(_) => { + report.failures += 1; + tracing::error!( + tick = %tick, + worker = registered.action.name(), + timeout_seconds = action_timeout.as_secs(), + "reconcile action timed out" + ); + } + } + } + + 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 { + actions_taken: usize, + failures: usize, +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use std::collections::HashMap; + use std::io; + + use tokio::sync::{Mutex, Notify}; + + 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, + } + + #[derive(Clone, Debug, Default)] + struct StalledAction { + started: Arc, + } + + impl Action for StalledAction { + fn name(&self) -> &'static str { + "stalled" + } + + fn run<'a>(&'a self, _database: &'a Db) -> ActionFuture<'a> { + Box::pin(async move { + self.started.notify_one(); + std::future::pending().await + }) + } + } + + 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)); + + 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)); + + 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] + 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); + } + + #[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::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"); + } +}