rainier_framework/

scheduling.rs

//! Scheduling — the tasks and commands that need more than the scheduler crate
//! depends on.
//!
//! `rainier-scheduler` knows about the container and the cache and nothing
//! else, which is what keeps it usable without a queue or a console. The two
//! things people actually schedule — a queued job and a console command — are
//! adapters, and they live here because this is the crate that has both.
//!
//! ```ignore
//! // src/routes/console.rs
//! pub fn schedule(schedule: &mut Schedule) {
//!     schedule.job(PruneSessions).daily_at("03:00").without_overlapping(HALF_HOUR);
//!     schedule.command("app:seed").weekly_on(1, "04:00").on_one_server();
//!     schedule.call("heartbeat", |_| Box::pin(async { Ok(()) })).every_five_minutes();
//! }
//! ```

use std::sync::Arc;
use std::time::Duration;

use chrono::Utc;
use rainier_cache::LockManager;
use rainier_console::{exit, Arguments, Command, Console};
use rainier_container::Application;
use rainier_queue::{Job, QueueManager};
use rainier_scheduler::{Schedule, ScheduledTask, Task};
use rainier_support::{BoxFuture, Error, Result};

/// A [`Task`] that dispatches a queued job.
///
/// The scheduler's own work is then only "put it on the queue", which is what
/// you want: a scheduled task that runs for ten minutes is ten minutes the
/// scheduler is not looking at anything else, and a worker is the thing built
/// to run long work.
pub struct JobTask<J> {
    job: Arc<J>,
}

impl<J: Job + Clone> JobTask<J> {
    /// Dispatch `job` when the schedule fires.
    pub fn new(job: J) -> Self {
        Self { job: Arc::new(job) }
    }
}

impl<J: Job + Clone> Task for JobTask<J> {
    fn name(&self) -> String {
        // The job's registered name, which is stable across restarts and
        // machines — exactly what a lock key has to be.
        J::NAME.to_string()
    }

    fn run<'a>(&'a self, app: &'a Application) -> BoxFuture<'a, Result<()>> {
        Box::pin(async move {
            let queue = app.resolve::<QueueManager>()?;
            queue.dispatch((*self.job).clone()).await?;
            Ok(())
        })
    }
}

/// A [`Task`] that runs a console command.
///
/// Laravel's `$schedule->command('inspire')`. The command is resolved from the
/// console at run time, so a typo is a failure of that occurrence rather than
/// of the boot — which is the trade for being able to name a command the
/// scheduler crate cannot see.
pub struct CommandTask {
    name: String,
    argv: Vec<String>,
    console: Arc<Console>,
}

impl CommandTask {
    /// Run `name` from `console`, with `argv` after it.
    pub fn new(console: Arc<Console>, name: impl Into<String>, argv: Vec<String>) -> Self {
        Self { name: name.into(), argv, console }
    }
}

impl Task for CommandTask {
    fn name(&self) -> String {
        if self.argv.is_empty() {
            self.name.clone()
        } else {
            format!("{} {}", self.name, self.argv.join(" "))
        }
    }

    fn run<'a>(&'a self, app: &'a Application) -> BoxFuture<'a, Result<()>> {
        Box::pin(async move {
            let mut argv = vec![self.name.clone()];
            argv.extend(self.argv.iter().cloned());

            let code = self.console.run_argv(app, argv).await;
            if code == exit::SUCCESS {
                Ok(())
            } else {
                Err(Error::internal(format!("`{}` exited with {code}", self.name)))
            }
        })
    }
}

/// `job` and `command`, on top of [`Schedule`].
///
/// An extension trait rather than methods on `Schedule`, because the scheduler
/// crate would otherwise have to depend on the queue and the console — and then
/// an application with neither would compile both.
pub trait ScheduleExt {
    /// Dispatch a queued job on a schedule.
    fn job<J: Job + Clone>(&mut self, job: J) -> &mut ScheduledTask;

    /// Run a console command on a schedule.
    fn command(&mut self, console: Arc<Console>, name: impl Into<String>) -> &mut ScheduledTask;

    /// Run a console command with arguments.
    fn command_with(
        &mut self,
        console: Arc<Console>,
        name: impl Into<String>,
        argv: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut ScheduledTask;
}

impl ScheduleExt for Schedule {
    fn job<J: Job + Clone>(&mut self, job: J) -> &mut ScheduledTask {
        self.add(JobTask::new(job))
    }

    fn command(&mut self, console: Arc<Console>, name: impl Into<String>) -> &mut ScheduledTask {
        self.add(CommandTask::new(console, name, Vec::new()))
    }

    fn command_with(
        &mut self,
        console: Arc<Console>,
        name: impl Into<String>,
        argv: impl IntoIterator<Item = impl Into<String>>,
    ) -> &mut ScheduledTask {
        self.add(CommandTask::new(console, name, argv.into_iter().map(Into::into).collect()))
    }
}

/// `schedule:run` — run whatever is due this minute, then exit.
///
/// What a system cron entry calls:
///
/// ```cron
/// * * * * * cd /srv/app && ./app schedule:run >> /dev/null 2>&1
/// ```
#[derive(Debug, Default)]
pub struct ScheduleRunCommand;

#[async_trait::async_trait]
impl Command for ScheduleRunCommand {
    fn name(&self) -> &str {
        "schedule:run"
    }

    fn description(&self) -> &str {
        "Run every scheduled task that is due"
    }

    fn help(&self) -> Option<&str> {
        Some(
            "Usage:\n  schedule:run\n\n\
             Runs the tasks due this minute and exits. Call it once a minute\n\
             from cron, a systemd timer, or a Kubernetes CronJob.\n\n\
             Exits non-zero if any task failed, so a supervisor notices.",
        )
    }

    async fn handle(&self, _args: &Arguments, app: &Application) -> Result<i32> {
        let schedule = app.resolve::<Schedule>()?;
        let locks = app.resolve::<LockManager>()?;

        refuse_a_broken_schedule(&schedule)?;

        let summary = schedule.run_due(app, &locks, Utc::now()).await;

        if summary.considered() == 0 {
            println!("Nothing is due.");
            return Ok(exit::SUCCESS);
        }

        for name in &summary.ran {
            println!("Ran: {name}");
        }
        for (name, why) in &summary.skipped {
            println!("Skipped: {name} — {why}");
        }
        for (name, error) in &summary.failed {
            eprintln!("Failed: {name} — {error}");
        }

        Ok(if summary.has_failures() { exit::FAILURE } else { exit::SUCCESS })
    }
}

/// `schedule:work` — the same thing in a loop, for a container.
///
/// For a deployment with no cron: one process that wakes at the top of each
/// minute. Equivalent to `schedule:run` on a timer, and easier to supervise.
#[derive(Debug, Default)]
pub struct ScheduleWorkCommand;

#[async_trait::async_trait]
impl Command for ScheduleWorkCommand {
    fn name(&self) -> &str {
        "schedule:work"
    }

    fn description(&self) -> &str {
        "Run the scheduler in the foreground, once a minute"
    }

    fn help(&self) -> Option<&str> {
        Some(
            "Usage:\n  schedule:work [--once]\n\n\
             Options:\n  --once  Run the due tasks once and exit\n\n\
             For a deployment with no cron — a container, a systemd unit. One\n\
             process, supervised like any other.",
        )
    }

    async fn handle(&self, args: &Arguments, app: &Application) -> Result<i32> {
        let schedule = app.resolve::<Schedule>()?;
        let locks = app.resolve::<LockManager>()?;

        refuse_a_broken_schedule(&schedule)?;

        if args.flag("once") {
            let summary = schedule.run_due(app, &locks, Utc::now()).await;
            return Ok(if summary.has_failures() { exit::FAILURE } else { exit::SUCCESS });
        }

        println!("Scheduler running — {} task(s). Press Ctrl-C to stop.", schedule.len());

        loop {
            // Sleep to the top of the next minute rather than for sixty
            // seconds. Sleeping for a fixed interval drifts: each pass takes a
            // little longer than it slept, and after enough hours a task
            // scheduled for `:00` is running at `:01` and missing its minute
            // entirely.
            tokio::time::sleep(until_next_minute()).await;

            let summary = schedule.run_due(app, &locks, Utc::now()).await;
            if summary.considered() > 0 {
                tracing::info!(
                    ran = summary.ran.len(),
                    skipped = summary.skipped.len(),
                    failed = summary.failed.len(),
                    "scheduler pass"
                );
            }
        }
    }
}

/// `schedule:list` — what is scheduled, and when it next runs.
#[derive(Debug, Default)]
pub struct ScheduleListCommand;

#[async_trait::async_trait]
impl Command for ScheduleListCommand {
    fn name(&self) -> &str {
        "schedule:list"
    }

    fn description(&self) -> &str {
        "List every scheduled task and when it next runs"
    }

    async fn handle(&self, _args: &Arguments, app: &Application) -> Result<i32> {
        let schedule = app.resolve::<Schedule>()?;

        if schedule.is_empty() {
            println!("Nothing is scheduled.");
            return Ok(exit::SUCCESS);
        }

        let now = Utc::now();
        let name_width =
            schedule.tasks().iter().map(|t| t.description().len()).max().unwrap_or(4).max(4);
        let expression_width =
            schedule.tasks().iter().map(|t| t.expression().source().len()).max().unwrap_or(10);

        println!(
            "{:name_width$}  {:expression_width$}  {:19}  GUARDS",
            "TASK", "SCHEDULE", "NEXT RUN (UTC)"
        );

        for task in schedule.tasks() {
            let next = task
                .next_run(now)
                .map(|at| at.format("%Y-%m-%d %H:%M:%S").to_string())
                .unwrap_or_else(|| "never".to_string());

            let mut guards = Vec::new();
            if task.overlap_ttl().is_some() {
                guards.push("without-overlapping");
            }
            if task.is_one_server() {
                guards.push("one-server");
            }

            println!(
                "{:name_width$}  {:expression_width$}  {next:19}  {}",
                task.description(),
                task.expression().source(),
                guards.join(", "),
            );
        }

        // Both of these make a lock silently useless, so they are worth
        // printing even when nobody asked.
        for (name, error) in schedule.errors() {
            eprintln!("\nInvalid schedule for `{name}`: {error}");
        }
        for name in schedule.duplicate_names() {
            eprintln!(
                "\nTwo tasks are both called `{name}`. They share a lock, so \
                 `without_overlapping` between them makes each block the other."
            );
        }

        println!("\n{} task(s)", schedule.len());
        Ok(exit::SUCCESS)
    }
}

/// Fail rather than run a schedule with a task on the wrong expression.
///
/// The builders are lenient so a chain does not need a `?` in the middle of it;
/// this is where that leniency is paid back. A task quietly on `* * * * *`
/// because `.daily_at("3pm")` did not parse is worse than a scheduler that
/// refuses to start.
fn refuse_a_broken_schedule(schedule: &Schedule) -> Result<()> {
    let errors = schedule.errors();
    if errors.is_empty() {
        return Ok(());
    }

    let listed: Vec<String> =
        errors.iter().map(|(name, error)| format!("`{name}`: {error}")).collect();

    Err(Error::internal(format!(
        "the schedule has {} task(s) with an invalid expression — {}",
        errors.len(),
        listed.join("; ")
    )))
}

/// How long until the top of the next minute.
fn until_next_minute() -> Duration {
    let now = Utc::now();
    let seconds = 60 - (now.timestamp() % 60);

    // At least a second, so a pass that finishes instantly cannot spin through
    // the same minute twice.
    Duration::from_secs(seconds.max(1) as u64)
}

#[cfg(test)]
mod tests {
    use super::*;
    use rainier_cache::MemoryCache;
    use rainier_console::Console;

    fn app_with(schedule: Schedule) -> Application {
        let app = Application::new(".");
        app.instance(schedule);
        app.instance(LockManager::new(Arc::new(MemoryCache::new())));
        app
    }

    #[tokio::test]
    async fn schedule_run_reports_what_it_did() {
        let mut schedule = Schedule::new();
        schedule.call("tick", |_| Box::pin(async { Ok(()) }));

        let app = app_with(schedule);
        let console = Console::new("rainier").register(ScheduleRunCommand);

        assert_eq!(console.run_argv(&app, ["schedule:run"]).await, exit::SUCCESS);
    }

    #[tokio::test]
    async fn schedule_run_exits_non_zero_when_a_task_fails() {
        // So a supervisor, or the `&&` in a deploy script, notices.
        let mut schedule = Schedule::new();
        schedule.call("broken", |_| Box::pin(async { Err(Error::internal("boom")) }));

        let app = app_with(schedule);
        let console = Console::new("rainier").register(ScheduleRunCommand);

        assert_eq!(console.run_argv(&app, ["schedule:run"]).await, exit::FAILURE);
    }

    #[tokio::test]
    async fn a_schedule_with_an_unparsable_expression_refuses_to_run() {
        // The builders are lenient mid-chain; this is where that is paid back.
        let mut schedule = Schedule::new();
        schedule.call("wrong", |_| Box::pin(async { Ok(()) })).cron("not a cron expression");

        let app = app_with(schedule);
        let console = Console::new("rainier").register(ScheduleRunCommand);

        assert_eq!(console.run_argv(&app, ["schedule:run"]).await, exit::FAILURE);
    }

    #[tokio::test]
    async fn schedule_list_names_a_duplicate() {
        let mut schedule = Schedule::new();
        schedule.call("prune", |_| Box::pin(async { Ok(()) })).hourly();
        schedule.call("prune", |_| Box::pin(async { Ok(()) })).daily();

        let app = app_with(schedule);
        let console = Console::new("rainier").register(ScheduleListCommand);

        // It prints rather than fails — a duplicate is a warning, not a reason
        // to stop the whole schedule.
        assert_eq!(console.run_argv(&app, ["schedule:list"]).await, exit::SUCCESS);
        assert_eq!(app.resolve::<Schedule>().unwrap().duplicate_names(), vec!["prune"]);
    }

    #[tokio::test]
    async fn schedule_work_once_runs_a_single_pass() {
        let mut schedule = Schedule::new();
        schedule.call("tick", |_| Box::pin(async { Ok(()) }));

        let app = app_with(schedule);
        let console = Console::new("rainier").register(ScheduleWorkCommand);

        assert_eq!(console.run_argv(&app, ["schedule:work", "--once"]).await, exit::SUCCESS);
    }

    #[test]
    fn the_sleep_lands_on_the_next_minute_rather_than_drifting() {
        // Sleeping a fixed sixty seconds drifts by however long each pass took,
        // and after enough hours a task scheduled for `:00` runs at `:01`.
        let wait = until_next_minute();

        assert!(wait.as_secs() >= 1 && wait.as_secs() <= 60, "{wait:?}");
    }

    #[test]
    fn a_broken_schedule_is_described_rather_than_just_refused() {
        let mut schedule = Schedule::new();
        schedule.call("wrong", |_| Box::pin(async { Ok(()) })).cron("* * *");

        let err = refuse_a_broken_schedule(&schedule).unwrap_err();
        assert!(err.message().contains("`wrong`"), "{}", err.message());
        assert!(err.message().contains("5"), "{}", err.message());
    }
}