Add reconcile loop scheduler (#69)

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.

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<dyn Error + Send + Sync>;
+pub type ActionFuture<'a> =
+    Pin<Box<dyn Future<Output = Result<Vec<Outcome>, 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<String>, action: impl Into<String>) -> 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<dyn Action>,
+}
+
+#[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<Duration>,
+    actions: Vec<RegisteredAction>,
+}
+
+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<A>(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<bool>) {
+        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<Self>, tick: Tick, mut shutdown: watch::Receiver<bool>) {
+        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<Mutex<HashMap<i64, String>>>,
+    }
+
+    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<Notify>,
+    }
+
+    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");
+    }
+}