Crate oxana

Expand description

§Oxana

Oxana is a Redis-backed job processing library for Rust. It powers the background job infrastructure behind Player.gg and Firstlook.gg, serving hundreds of studios and millions of players.

Oxana focuses on simplicity and depth over breadth - one backend, done well.

Oxana Web Dashboard

§Key Features

Isolated Queues - separate queues with independent concurrency and configuration
Retries - automatic retry with configurable backoff
Scheduled Jobs - run jobs at specific times or after delays
Cron Jobs - periodic jobs using cron expressions
Batch Processing - process jobs together with size and timeout based batching
Dynamic Queues - create and manage queues at runtime
Runtime Queue Controls - pause queues and adjust dynamic concurrency without restarting workers
Throttling - rate-limit job processing per queue
Unique Jobs - deduplicate jobs so only one instance runs at a time
Resumable Jobs - resume from where a job left off on retry
Resilient Jobs - survive worker crashes and restarts
Graceful Shutdown - clean shutdown with in-progress job handling
Web Dashboard - built-in UI for monitoring jobs, queues, metrics, and cron - pure Rust, no JS toolchain
Prometheus Metrics - export queue and job metrics for monitoring
Well Tested - comprehensive integration test suite

§Quick Start

cargo add oxana@2.1.3 --features registry

The registry feature (not enabled by default) powers #[derive(oxana::Registry)] and runtime.register::<...>() used below; without it, register queues and workers explicitly with runtime.queue::<...>() and runtime.worker::<...>().

use oxana::Storage;
use serde::{Serialize, Deserialize};

#[derive(oxana::Registry)]
struct ComponentRegistry(oxana::ComponentRegistry<MyContext>);

#[derive(Debug, thiserror::Error)]
enum MyError {}

#[derive(Debug, Clone)]
struct MyContext {}

#[derive(Debug, Serialize, Deserialize, oxana::Job)]
struct MyJob {
    data: String,
}

#[derive(oxana::Worker)]
#[oxana(context = MyContext)]
struct MyWorker;

impl MyWorker {
    async fn process(&self, job: MyJob, _ctx: &oxana::JobContext) -> Result<(), MyError> {
        println!("Processing: {}", job.data);
        Ok(())
    }
}

#[derive(oxana::Queue)]
#[oxana(key = "my_queue", concurrency = 2)]
struct MyQueue;

#[tokio::main]
async fn main() -> Result<(), oxana::OxanaError> {
    let storage = Storage::from_env()?;
    let runtime = storage
        .runtime(MyContext {})
        .register::<ComponentRegistry>();

    storage.enqueue(MyQueue, MyJob { data: "hello".into() }).await?;
    runtime.run().await?;
    Ok(())
}

For more detailed usage examples, check out the examples directory.

§Web Dashboard

The oxana-web crate provides a built-in dashboard for monitoring jobs, queues, worker metrics, and cron schedules. It integrates as a nested axum router.

use oxana_web::OxanaWebState;

let runtime = storage
    .runtime(MyContext {})
    .register::<ComponentRegistry>();
let catalog = runtime.catalog();

let oxana_router = oxana_web::router(OxanaWebState::new(
    storage.clone(),
    catalog,
    "/oxana".to_string(),
));

let app = your_app_router().nest("/oxana", oxana_router);

The dashboard exposes these pages:

/ - Overview with job stats
/busy - Currently processing jobs
/queues - All queues with stats
/queues/{queue_key} - Jobs in a specific queue
/jobs/{job_id} - Details for a specific job
/metrics - Worker execution metrics
/metrics/job?worker=... - Metrics for a specific worker
/cron - Cron job schedules
/on-demand - Manually enqueue registered on-demand jobs
/scheduled - Scheduled jobs
/retries - Jobs pending retry
/dead - Dead letter queue

It also provides management actions for pausing and unpausing queues, changing dynamic queue concurrency from queue detail pages, wiping queues, and deleting individual jobs.

§Core Concepts

§Jobs and Workers

Jobs carry the data that gets enqueued and define enqueue-time metadata. Workers define the processing logic for jobs. Use the #[derive(oxana::Job)] and #[derive(oxana::Worker)] macros or implement the traits manually.

`Job` attributes (enqueue-time)	`Worker` attributes (execution-time)
`#[oxana(unique_id = "worker_{id}")]` - define unique job identifiers	`#[oxana(job = MyJob)]` - override inferred `FooWorker` -> `FooJob` binding
`#[oxana(on_conflict = Skip)]` - handle unique job conflicts (Skip or Replace)	`#[oxana(context = MyContext)]` - set worker context type
`#[oxana(resurrect = false)]` - disable crash resurrection for this job type	`#[oxana(error = MyError)]` - set worker error type
`#[oxana(resume = false)]` - reset prior-attempt job state on retry	`#[oxana(registry = MyRegistry)]` - choose component registry
`#[oxana(throttle_cost = 2)]` - set per-job throttle cost	`#[oxana(max_retries = 3)]` - set maximum retry attempts
`#[oxana(on_demand)]` - expose the job in the web dashboard for manual enqueueing	`#[oxana(retry_delay = 5)]` - set retry delay in seconds
	`#[oxana(cron(schedule = "/5 * * * *", queue = MyQueue))]` - schedule periodic jobs
	`#[oxana(batch_size = 100, batch_timeout_ms = 500)]` - process jobs in batches

Cron jobs may define a unique_id to prevent overlapping occurrences. Their conflict strategy must be Skip; on_conflict = Replace is rejected because replacing an in-flight occurrence can corrupt the following occurrence.

For job hooks, Self::... resolves to the job type. For worker hooks, Self::... resolves to the worker type.

On-demand argument templates infer editable placeholders from field types. Numeric primitives and common numeric ID newtypes named *Id or *ID are prefilled with 0.

Batch workers use all-or-nothing result semantics: if process_batch returns Ok(()), every job in the batch is marked successful; if it returns an error or panics, every job in that batch follows the normal retry or failure path. Batch handlers should therefore be idempotent, or should only commit external side effects after the whole batch is ready to succeed.

§Queues

Queues are the channels through which jobs flow. Use the #[derive(oxana::Queue)] macro or implement the Queue trait manually. Static queues do not need Serialize; dynamic queues need serializable fields so Oxana can derive each runtime queue key.

Queues can be:

Static: Defined at compile time with a fixed key
Dynamic: Created at runtime with each instance being a separate queue (requires struct fields)

Queue attributes:

#[oxana(key = "my_queue")] - Set static queue key
#[oxana(prefix = "dynamic")] - Set prefix for dynamic queues
#[oxana(concurrency = 2)] - Set a fixed concurrency limit
#[oxana(concurrency = Dynamic(2))] - Set a runtime-adjustable concurrency limit with default 2
#[oxana(throttle(window_ms = 2000, limit = 5))] - Configure throttling
#[oxana(discovery_interval_ms = 250)] - Set how often new dynamic queues are discovered (dynamic queues only)

§Component Registry

The component registry automatically discovers and registers all workers and queues in your application. Use #[derive(oxana::Registry)] to create a registry and storage.runtime(ctx).register::<ComponentRegistry>() to register them on a typed runtime.

§Storage

Storage provides the interface for job persistence - enqueueing, scheduling, state management, and queue monitoring.

Build it with Storage::from_env() or Storage::builder().build_from_env(), which read the REDIS_URL environment variable. Set REDIS_STATS_URL to store counters and metrics in a separate Redis instance; when it is not set, stats use REDIS_URL. Call storage.runtime(ctx) to create a typed worker runtime, register queues and workers on that runtime, then call runtime.run().await.

§Context

The context provides shared state and utilities to workers. It can include:

Database connections
Configuration
Shared resources
Job state (for resumable jobs)

Workers can persist job state with ctx.state.update(...) and read the state from the current attempt with ctx.state.get::<T>(). This is useful for resumable jobs that need to continue from the last completed item after a retry. Jobs resume state by default; use #[oxana(resume = false)] to clear prior-attempt state before each retry.

For long-running jobs, use ctx.state.update_progress(...) to store progress in a structured format. The web dashboard renders a progress bar when total is set:

ctx.state
    .update_progress((cursor, total, Some("importing users".to_string())))
    .await?;

update_progress accepts a cursor value, (cursor, total), or (cursor, total, note). Cursor-only state remains useful for resumable jobs and is shown as normal state in the dashboard. ctx.state.progress().await? reloads the latest stored progress for the current job.

§Runtime Configuration

Runtime configuration is done on the typed runtime builder, which allows you to:

Automatically register queues and workers via the component registry
Set up graceful shutdown
Configure exit conditions and runtime timing/backoff knobs
Customize how worker errors are stored on retry and dead jobs

Storage remains the enqueueing and monitoring handle; RuntimeBuilder<C> is the worker setup and execution handle for app context type C.

Worker errors use their Debug representation by default so error types that capture backtraces can include them in the dashboard. Applications can override the stored representation when needed:

let runtime = storage
    .runtime(ctx)
    .error_formatter(|error| error.to_string());

Backtraces must be enabled before the process starts (for example with RUST_BACKTRACE=1) and captured by the application’s error type. Diagnostic output may contain sensitive details, so only enable backtraces where access to stored job errors is appropriately restricted.

§Error Handling

Oxana uses a custom OxanaError type that covers all library error cases. Workers can define their own error type that implements std::error::Error.

§Prometheus Metrics

Enable the prometheus feature to expose metrics:

let metrics = storage.metrics().await?;
let output = metrics.encode_to_string()?;
// Serve `output` on your metrics endpoint

§Comparison with Similar Libraries

Feature	Oxana	Apalis	rusty-sidekiq	Fang
Backend	Redis	Redis, Postgres, `SQLite`, `MySQL`, AMQP, NATS	Redis	Postgres, `SQLite`, `MySQL`
Retries	Yes	Yes (tower layer)	Yes	Yes
Scheduled Jobs	Yes	Yes	Yes	Yes
Cron	Yes	Yes	Yes	Yes
Unique Jobs	Yes	No	Yes	Yes
Throttling	Yes	No	No	No
Dynamic Queues	Yes	No	No	No
Resumable Jobs	Yes	No	No	No
Graceful Shutdown	Yes	Yes	Partial	No
Web UI	Yes	Yes (apalis-board)	No (uses Ruby Sidekiq UI)	No
License	MIT	MIT	MIT	MIT

Oxana focuses on depth with a single Redis backend rather than breadth across multiple backends. It is the only Rust job library offering resumable jobs and combines unique jobs, throttling, and a built-in web dashboard in one package.

Apalis offers the most backend options and integrates with the tower middleware ecosystem, making it highly extensible. It suits projects that need backend flexibility or already use tower layers. However, its breadth of abstraction can come at the cost of reliability and debuggability in production.

rusty-sidekiq is wire-compatible with Ruby Sidekiq, making it ideal for teams migrating from or coexisting with Ruby services. It can share queues with Ruby Sidekiq workers and use the existing Sidekiq web UI.

Fang is SQL-database-backed (no Redis dependency) with both async and threaded execution modes. A good fit for projects that prefer Postgres/SQLite over Redis.

§License

Oxana is licensed under the MIT License.

Structs§

BatchItem
BoxError: Type-erased worker error, used as the default Worker::Error type.
Catalog: Catalog of all registered workers and queues.
CronWorkerInfo: Information about a registered cron worker.
DrainStats
DynamicQueueStats: Statistics for a dynamic sub-queue.
JobContext
JobData
JobEnvelope
JobMeta
JobMetricsDetail
JobMetricsHistogramBucket
JobMetricsPoint
JobMetricsQuery
JobMetricsSnapshot
JobMetricsTotals
JobProgress
JobProgressIterator
JobState
MetricIdentity
OnDemandJobInfo: Information about an on-demand job exposed in the web dashboard.
OnDemandJobRegistration
Process: Information about an Oxana worker process.
QueueConfig
QueueInfo: Information about a registered queue.
QueueLengthMetricsPoint
QueueLengthMetricsSeries
QueueLengthMetricsSnapshot
QueueListOpts: Options for listing jobs in a queue.
QueueRateStats: Historical per-queue rate estimates.
QueueRuntimeConfig
QueueStats: Statistics for a specific queue.
QueueThrottle
QueueThrottleInfo: Throttle configuration for a queue.
RunStats
RuntimeBuilder
Stats: Overall statistics for the Oxana job queue system.
StatsGlobal: Global aggregate statistics.
StatsProcessing: Information about a job currently being processed.
Storage: Storage provides the main interface for job management in Oxana.
StorageBuilder
StorageBuilderTimeouts
WorkerBatchConfig
WorkerConfig
WorkerInfo: Information about a registered worker.
WorkerMetricsSummary

Enums§

JobConflictStrategy
OxanaError
QueueConcurrency
QueueKind
QueueState
WorkerConfigKind

Constants§

HISTOGRAM_BUCKET_INTERVALS_MS: Maximum execution time represented by each histogram bucket.
HISTOGRAM_BUCKET_LABELS: Display labels for HISTOGRAM_BUCKET_INTERVALS_MS.

Traits§

FromContext
IntoWorkerError
Job
Queue
UniqueJobId: Resolves the deterministic ID used by a unique job.
Worker

Functions§

job_batch_factory
job_envelope_factory
job_factory

Type Aliases§

JobId

Derive Macros§

Job: Generates impl for oxana::Job.
Queue: Generates impl for oxana::Queue.
Registry: Helper to define a component registry.
Worker: Generates impl for oxana::Worker.

Crate oxana

Crate oxana Copy item path

§Oxana

§Key Features

§Quick Start

§Web Dashboard

§Core Concepts

§Jobs and Workers

§Queues

§Component Registry

§Storage

§Context

§Runtime Configuration

§Error Handling

§Prometheus Metrics

§Comparison with Similar Libraries

§License

Structs§

Enums§

Constants§

Traits§

Functions§

Type Aliases§

Derive Macros§

Crate oxana