RunAt

A distributed job scheduler for Rust with PostgreSQL backend support.

Features

• Distributed Job Scheduling: Run jobs across multiple workers with PostgreSQL-backed coordination
• Cron Support: Schedule recurring jobs using cron expressions
• Retry Mechanisms: Built-in exponential backoff and custom retry strategies
• Type-Safe Jobs: Leverage Rust's type system for job definitions
• Async/Await: Built on Tokio for efficient async job execution
• Macro Support: Convenient procedural macros for defining jobs

Installation

Add this to your Cargo.toml:

[dependencies]

runat = "0.1.1"

For PostgreSQL support (enabled by default):

[dependencies]

runat = { version = "0.1.1", features = ["postgres"] }

Optional features:

• postgres - PostgreSQL backend (enabled by default)
• tracing - Tracing support for observability

Quick Start

Basic Job

use runat::{BackgroundJob, JobQueue, PostgresDatastore, Result, job};
use sqlx::postgres::PgPoolOptions;
use std::sync::Arc;

#[job]
async fn send_email(to: String, subject: String, body: String) -> Result<()> {
    println!("Sending email to {}: {}", to, subject);
    // Your email sending logic here
    Ok(())
}

#[tokio::main]
async fn main() -> Result<()> {
    // Connect to PostgreSQL
    let database_url = std::env::var("DATABASE_URL")
        .unwrap_or_else(|_| "postgres://user:pass@localhost/db".to_string());

    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&database_url)
        .await?;

    // Initialize datastore and run migrations
    let datastore = PostgresDatastore::new(pool).await?;
    datastore.migrate().await?;

    // Create job queue (JobQueue is Clone, so no Arc needed)
    let queue = JobQueue::with_datastore(Arc::new(datastore));

    // Enqueue a job
    queue.enqueue(
        JobSendEmail {
            to: "user@example.com".to_string(),
            subject: "Welcome!".to_string(),
            body: "Thanks for signing up!".to_string(),
        }.job()?
    ).await?;

    Ok(())
}

Scheduled Jobs with Cron

// Schedule a job to run every 10 seconds
queue.enqueue(
    JobSendEmail {
        to: "admin@example.com".to_string(),
        subject: "Daily Report".to_string(),
        body: "Here's your daily report".to_string(),
    }
    .job()?
    .cron("*/10 * * * * * *")?
).await?;

Custom Job Implementation

use runat::{Job, Executable, Result};
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
struct ProcessPayment {
    user_id: String,
    amount: f64,
}

#[async_trait::async_trait]
impl Executable for ProcessPayment {
    type Output = Result<String>;

    async fn execute(&mut self) -> Self::Output {
        println!("Processing ${} payment for user {}", self.amount, self.user_id);
        // Your payment processing logic here
        Ok(format!("Payment processed for {}", self.user_id))
    }
}

Running Workers

You can run workers in two ways:

Option 1: Run worker directly from the queue (recommended)

// JobQueue is Clone, so you can clone it before moving into the task
let queue_clone = queue.clone();
tokio::spawn(async move {
    queue_clone.start_worker().await
});

Option 2: Create a worker instance

use runat::JobWorker;

// Create a worker from the queue
let worker = queue.worker();

// Run the worker
tokio::spawn(async move {
    worker.run().await
});

Retry Strategies

use runat::Retry;
use chrono::Duration;

// Create a job with exponential backoff retry
let mut job = Job::new(email_job)?;
job.max_attempts = 5;
job = job.retry(Retry::exponential(Duration::seconds(5)));

queue.enqueue(job).await?;

Running Tests

cargo test