apalis

Simple and extensible multithreaded background task and messages processing library for Rust

Features

• Simple and predictable task handling - Task handlers are just async functions with a macro-free API
• Robust task execution - Built-in support for retries, timeouts, and error handling
• Multiple storage backends - Support for Redis, PostgreSQL, SQLite, and in-memory storage
• Advanced task management - Task prioritization, scheduling, metadata, and result tracking
• Scalable by design - Distributed backends with configurable concurrency and multi-threaded execution
• Familiar dependency injection - Similar to popular frameworks like actix and axum
• Runtime agnostic - Works with tokio, async-std, and other async runtimes
• Production ready - Built-in monitoring, metrics, graceful shutdown, and comprehensive error reporting
• Extensible middleware system - Take full advantage of the tower ecosystem of services and utilities
• Optional web interface - Manage and monitor your tasks through a web UI

Crate ecosystem

Source	Crate	Examples
apalis-cron
apalis-redis
apalis-sqlite
apalis-postgres
apalis-mysql
apalis-amqp
apalis-core
apalis-workflow
apalis-board
apalis-board-api

Getting Started

To get started, just add to Cargo.toml

[dependencies]
apalis = { version = "1.0.0-beta.1" }
# apalis-redis = { version = "1.0.0-alpha.1" } # Use redis/sqlite/postgres etc

Usage

use apalis::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Serialize, Clone)]
struct Email {
    to: String,
}

/// A function called for every task
async fn send_email(task: Email, data: Data<usize>) {
  // execute task
}


#[tokio::main]
async fn main() {
    let mut storage = MemoryStorage::new();
     storage
        .push(Email {
            to: "test@example.com".to_string(),
        })
        .await
        .expect("Could not add tasks");

    WorkerBuilder::new("email-worker")
      .backend(storage)
      .concurrency(2)
      .parallelize(tokio::spawn)
      .data(0usize)
      .build(send_email)
      .run()
      .await;
}

Workflows

• apalis has first class support for sequential, dag and conditional workflows.

use apalis::prelude::*;
use apalis_workflow::*;
use std::time::Duration;
use apalis_core::backend::json::JsonStorage;

#[tokio::main]
async fn main() {
   let workflow = Workflow::new("odd-numbers-workflow")
       .delay_for(Duration::from_millis(1000))
       .and_then(|a: usize| async move { Ok::<_, BoxDynError>((0..a).collect::<Vec<_>>()) })
       .filter_map(|x| async move { if x % 2 != 0 { Some(x) } else { None } })
       .and_then(|a: Vec<usize>| async move {
           println!("Sum: {}", a.iter().sum::<usize>());
           Ok::<_, BoxDynError>(())
        });

   let mut in_memory = JsonStorage::new_temp().unwrap();

   in_memory.push_start(10).await.unwrap();

   let worker = WorkerBuilder::new("rango-tango")
       .backend(in_memory)
       .on_event(|ctx, ev| {
           println!("On Event = {:?}", ev);
       })
       .build(workflow);
   worker.run().await.unwrap();
}

For more functionality like fold, filter_map and other combinators checkout the docs

Feature flags

• full - All the available features
• tracing (enabled by default) — Support Tracing 👀
• sentry — Support for Sentry exception and performance monitoring
• prometheus — Support Prometheus metrics
• retry — Support direct retrying tasks
• timeout — Support timeouts on tasks
• limit — Support for concurrency and rate-limiting
• filter — Support filtering tasks based on a predicate
• catch-panic - Catch panics that occur during execution

How apalis works

Here is a basic example of how the core parts integrate

sequenceDiagram
    participant App
    participant Worker
    participant Backend

    App->>+Backend: Add task to queue
    Backend-->>+Worker: Job data
    Worker->>+Backend: Update task status to 'Running'
    Worker->>+App: Started task
    loop task execution
        Worker-->>-App: Report task progress
    end
    Worker->>+Backend: Update task status to 'completed'

Loading

Observability

With the web UI, you can manage your jobs through a simple interface. Check out this working example to see how it works.

Integrations

• zino: Next-generation framework for composable applications in Rust.
• spring-rs: Application framework written in Rust, inspired by Java's SpringBoot

Projects using apalis

• Ryot: A self hosted platform for tracking various facets of your life - media, fitness etc.
• Hyprnote: Local-first AI Notepad for Private Meetings
• oxy: A framework for building SQL agents and automations for analytics.
• OpenZeppelin Relayer: OpenZeppelin relayer service
• Summarizer: Podcast summarizer
• Universal Inbox: Universal Inbox is a solution that centralizes all your notifications and tasks in one place to create a unique inbox.
• Hatsu: Self-hosted and fully-automated ActivityPub bridge for static sites.
• Stamon: A lightweight self-hosted status monitoring tool
• Gq: open-source filtering tool for JSON and YAML files

External examples

• Shuttle: Using apalis-cron with shuttle.rs
• Actix-Web: Using apalis-redis with actix-web
• Zino-Example: Using zino

Resources

• Background job processing with rust using actix and redis
• Feasibility of implementing cronjob with Rust programming language
• How to schedule and run cron jobs in Rust using apalis

Thanks to

• tower - Tower is a library of modular and reusable components for building robust networking clients and servers.
• redis-rs - Redis library for rust
• sqlx - The Rust SQL Toolkit
• cron - A cron expression parser and schedule explorer

Contributing

Please read CONTRIBUTING.md for details on our code of conduct, and the process for submitting pull requests to us.

Versioning

We use SemVer for versioning. For the versions available, see the tags on this repository.

Contributors

See also the list of contributors who participated in this project.

License

This project is licensed under the MIT License - see the LICENSE.md file for details