Crate apalis_sqlite 

apalis-sqlite

Background task processing for Rust, powered by apalis and SQLite.

---

Features

• Reliable job queue — SQLite-backed persistence survives process restarts; no external broker needed.
• Durable execution — jobs are written to disk before processing begins, so a crash mid-flight won’t silently drop work.
• Automatic retries — configure max_attempts per task; failed jobs are re-enqueued with backoff and never lost.
• Heartbeat & orphan recovery — workers emit periodic heartbeats; jobs held by a dead worker are automatically re-enqueued.
• Scheduled & delayed jobs — use run_after to enqueue work that should only execute at or after a future point in time.
• Priority queues — assign integer priorities so high-urgency jobs are always picked up first.
• Multiple polling strategies — choose between standard polling and event-driven (hooked) storage to trade latency for CPU usage.
• Multi-step workflows — chain async steps into pipelines with apalis-workflow; each stage only runs if the previous one succeeds.
• Shared storage — multiple job types over a single SQLite connection.
• Custom codecs — pluggable serialization/deserialization of job payloads as raw bytes.
• First-class apalis integration — works seamlessly with workers, tower layers, and middleware.

---

Examples

Basic Worker

Set up a pool, push jobs, and run a worker:

use std::time::Duration;

use apalis::prelude::*;
use apalis_sqlite::*;
use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() {
    let pool = SqlitePool::connect(":memory:").await.unwrap();
    SqliteStorage::setup(&pool).await.unwrap();
    let mut backend = SqliteStorage::new(&pool);

    let mut start = 0;
    let mut items = stream::repeat_with(move || {
        start += 1;
        TaskBuilder::new(start)
            .run_after(Duration::from_secs(1))
            .priority(1)
            .max_attempts(5)
            .build()
    })
    .take(10);
    backend.push_all(&mut items).await.unwrap();

    async fn send_reminder(item: usize, wrk: WorkerContext) -> Result<(), BoxDynError> {
        Ok(())
    }

    let worker = WorkerBuilder::new("worker-1")
        .backend(backend)
        .build(send_reminder);
    worker.run().await.unwrap();
}

Event-Driven Worker (Hooked)

Use SQLite update hooks to react to new jobs with minimal latency:

use std::time::Duration;

use apalis::prelude::*;
use apalis_sqlite::*;
use futures::stream::{self, StreamExt};

#[tokio::main]
async fn main() {
    let (pool, listener) = SqliteStorage::connect_with_callback(":memory:").unwrap();

    let config = Config::default()
        .queue("queue")
        .batch_size(5);

    SqliteStorage::setup(&pool).await.unwrap();

    let strategy = Strategy::new()
        .stream(listener.ready_chunks(config.batch_size))
        .interval(Duration::from_millis(500));
    

    tokio::spawn({
        let pool = pool.clone();
        let config = config.clone();
        let queue = config.queue.to_string();
        async move {
            tokio::time::sleep(Duration::from_secs(2)).await;
            let mut start = 0;
            let items = stream::repeat_with(move || {
                start += 1;
                TaskBuilder::new(serde_json::to_vec(&start).unwrap())
                    .run_after(Duration::from_secs(1))
                    .priority(start)
                    .build()
            })
            .take(20)
            .collect::<Vec<_>>()
            .await;
            let mut conn = pool.acquire().await.unwrap();
            apalis_sqlite::queries::push_tasks(&mut conn, &queue, &items).await.unwrap();
        }
    });

    let backend = SqliteStorage::new(&pool)
        .with_config(config)
        .poll_with_strategy(strategy);

    async fn send_reminder(item: usize, wrk: WorkerContext) -> Result<(), BoxDynError> {
        Ok(())
    }

    let worker = WorkerBuilder::new("worker-2")
        .backend(backend)
        .build(send_reminder);
    worker.run().await.unwrap();
}

Sequential Workflow (Order Fulfillment Pipeline)

Chain async steps to model a real-world order processing pipeline — validating payment, reserving inventory, dispatching a shipment notification, and emailing the customer. Each stage only runs if the previous one succeeds; a failure at any step returns an error and the job can be retried from the beginning.

use std::time::Duration;

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

#[derive(Debug, Clone, Serialize, Deserialize)]
struct Order {
    id: u64,
    customer_email: String,
    items: Vec<String>,
    total_cents: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct ChargedOrder {
    order_id: u64,
    customer_email: String,
    transaction_id: String,
    items: Vec<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct StockedItem {
    order_id: u64,
    sku: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct DispatchedOrder {
    order_id: u64,
    customer_email: String,
    tracking_number: String,
    item_count: usize,
}

/// Simulates an inventory check — returns `None` for out-of-stock SKUs so
async fn check_stock(item: StockedItem) -> Option<StockedItem> {
    let out_of_stock = ["WIDGET-C", "WIDGET-D"];
    if out_of_stock.contains(&item.sku.as_str()) {
        eprintln!("[{}] {} is out of stock — skipping", item.order_id, item.sku);
        None
    } else {
        Some(item)
    }
}

#[tokio::main]
async fn main() {
    let workflow = SteppedFlow::new("order-fulfillment")
        // Step 1: Charge payment and return a record of the charged order.
        .and_then(|order: Order| async move {
            println!(
                "[{}] Charging ${:.2} for {} item(s)...",
                order.id,
                order.total_cents as f64 / 100.0,
                order.items.len(),
            );
            // Call your payment provider (e.g. Stripe) here.
            Ok::<ChargedOrder, BoxDynError>(ChargedOrder {
                order_id: order.id,
                customer_email: order.customer_email,
                transaction_id: format!("txn_{}", order.id),
                items: order.items,
            })
        })
        .and_then(|charged: ChargedOrder| async move {
            let stocked = charged
                .items
                .into_iter()
                .map(|sku| StockedItem { order_id: charged.order_id, sku })
                .collect::<Vec<_>>();
            Ok::<(ChargedOrder, Vec<StockedItem>), BoxDynError>((
                ChargedOrder { items: vec![], ..charged },
                stocked,
            ))
        })
        .and_then(|(charged, items): (ChargedOrder, Vec<StockedItem>)| async move {
            let available = futures::future::join_all(items.into_iter().map(check_stock))
                .await
                .into_iter()
                .flatten()
                .collect::<Vec<_>>();

            if available.is_empty() {
                return Err("No items available to fulfil this order".into());
            }

            println!(
                "[{}] {} item(s) confirmed in stock",
                charged.order_id,
                available.len()
            );
            Ok::<(ChargedOrder, Vec<StockedItem>), BoxDynError>((charged, available))
        })
        // Step 3: Brief pause before handing off to the courier API.
        .delay_for(Duration::from_millis(500))
        // Step 4: Dispatch the shipment and obtain a tracking number.
        .and_then(|(charged, items): (ChargedOrder, Vec<StockedItem>)| async move {
            let tracking_number = format!("TRACK-{}", charged.order_id);
            println!(
                "[{}] Dispatched {} item(s) — tracking: {}",
                charged.order_id,
                items.len(),
                tracking_number,
            );
            // Call your courier API (e.g. EasyPost, Shippo) here.
            Ok::<DispatchedOrder, BoxDynError>(DispatchedOrder {
                order_id: charged.order_id,
                customer_email: charged.customer_email,
                tracking_number,
                item_count: items.len(),
            })
        })
        // Step 5: Notify the customer by email.
        .and_then(|dispatched: DispatchedOrder| async move {
            println!(
                "[{}] Emailing {} — your {} item(s) are on the way! ({})",
                dispatched.order_id,
                dispatched.customer_email,
                dispatched.item_count,
                dispatched.tracking_number,
            );
            // Send a transactional email (e.g. Resend, SendGrid) here.
            Ok::<(), BoxDynError>(())
        });

    let pool = SqlitePool::connect(":memory:").await.unwrap();
    SqliteStorage::setup(&pool).await.unwrap();

    let config = Config::default().queue("order-fulfillment");
    let mut sqlite = SqliteStorage::new(&pool).with_config(config);

    sqlite
        .push_start(Order {
            id: 1001,
            customer_email: "alice@example.com".into(),
            items: vec!["WIDGET-A".into(), "WIDGET-B".into(), "WIDGET-C".into()],
            total_cents: 7499,
        })
        .await
        .unwrap();

    let worker = WorkerBuilder::new("fulfillment-worker")
        .backend(sqlite)
        .on_event(|ctx, ev| {
            println!("Event: {:?}", ev);
            if matches!(ev, Event::Error(_)) {
                ctx.stop().unwrap();
            }
        })
        .build(workflow);

    worker.run().await.unwrap();
}

DAG Workflow

Chain async steps to model a real-world ETL pipeline which runs multiple steps concurrently, then collects the results.

use std::time::Duration;

use apalis::prelude::*;
use apalis_codec::msgpack::MsgPackCodec;
use apalis_sqlite::SqliteStorage;
use apalis_workflow::*;
use sqlx::SqlitePool;

async fn get_name(user_id: u32) -> Result<String, BoxDynError> {
    Ok(user_id.to_string())
}

async fn get_age(user_id: u32) -> Result<usize, BoxDynError> {
    tokio::time::sleep(Duration::from_millis(800)).await;
    Ok(user_id as usize + 20)
}

async fn get_address(user_id: u32) -> Result<usize, BoxDynError> {
    tokio::time::sleep(Duration::from_millis(500)).await;
    Ok(user_id as usize + 100)
}

async fn collector(
    (name, age, address): (String, usize, usize),
    wrk: WorkerContext,
) -> Result<usize, BoxDynError> {
    let result = name.parse::<usize>()? + age + address;
    wrk.stop().unwrap();
    Ok(result)
}

#[tokio::main]
async fn main() {
    let pool = SqlitePool::connect(":memory:").await.unwrap();
    SqliteStorage::setup(&pool).await.unwrap();
    let mut backend = SqliteStorage::new(&pool).with_codec(MsgPackCodec);
    backend.start_fan_out(vec![42, 43, 44]).await.unwrap();

    let dag_flow = GraphFlow::new("user-etl-workflow");
    let get_name = dag_flow.add_task(get_name);
    let get_age = dag_flow.add_task(get_age);
    let get_address = dag_flow.add_task(get_address);
    dag_flow
        .add_task(collector)
        .depends_on((&get_name, &get_age, &get_address)); // Order and types matters here

    dag_flow.validate().unwrap(); // Ensure DAG is valid

    let worker = WorkerBuilder::new("rango-tango")
        .backend(backend)
        .on_event(|_c, e| {
            println!("{e:?},");
        })
        .build(dag_flow);
    worker.run().await.unwrap();
}

Shared Storage (Multiple Job Types)

Run multiple job types over a single SQLite connection:

use std::{collections::HashMap, time::Duration};

use apalis::prelude::*;
use apalis_sqlite::{shared::SqliteStorageFactory, SqliteStorage};
use futures::stream;

#[tokio::main]
async fn main() {
    let mut store = SqliteStorageFactory::new(":memory:");
    SqliteStorage::setup(store.pool()).await.unwrap();

    let mut map_store = store.create().unwrap();
    let mut int_store = store.create().unwrap();

    map_store
        .push_stream(&mut stream::iter(vec![HashMap::<String, String>::new()]))
        .await
        .unwrap();
    int_store.push(99).await.unwrap();

    async fn send_reminder<T>(
        _: T,
        _task_id: TaskId,
        wrk: WorkerContext,
    ) -> Result<(), BoxDynError> {
        tokio::time::sleep(Duration::from_secs(2)).await;
        wrk.stop().unwrap();
        println!("Reminder sent!");
        Ok(())
    }

    let int_worker = WorkerBuilder::new("rango-tango-2")
        .backend(int_store)
        .build(send_reminder);
    let map_worker = WorkerBuilder::new("rango-tango-1")
        .backend(map_store)
        .build(send_reminder);
    tokio::try_join!(int_worker.run(), map_worker.run()).unwrap();
}

---

Observability

Monitor and inspect your jobs visually using apalis-board:

---

Related Crates

• apalis — the core worker and middleware framework.
• apalis-workflow — multi-step workflow support built on apalis.
• apalis-board — web UI for monitoring jobs.

---

Showcase and Examples

• ryot - A self hosted platform for tracking various facets of your life - media, fitness and more.
• decomp.dev - A video game decompilation tool
• actix-ntfy-service - Basic example that shows how to publish notifications using apalis-sqlite

---

License

Licensed under either of:

• Apache License, Version 2.0
• MIT License

at your option.

Modules

callback

Callbacks allow realtime listeners for new jobs

queries

Queries needed for polling, updating and exposing tasks.

shared

Shared SQLite storage for multiple workers.

Structs

Config

Configuration for a worker’s queue, batching, and liveness detection.

JsonCodec

Json encoding and decoding

SqliteConnectOptions

Options and flags which can be used to configure a SQLite connection.

SqliteConnection

A connection to an open Sqlite database.

SqliteStorage

SqliteStorage is a storage backend for apalis using sqlite as the database.

Enums

Error

Represents a wrapper for errors encountered on this crate

SqlxError

Represents all the ways a method can fail within SQLx.

Type Aliases

SqlitePool

An alias for Pool, specialized for SQLite.

SqlitePoolOptions

An alias for PoolOptions, specialized for SQLite.

SqliteTask

An alias for Task specialized for Sqlite