| Crates.io | graphile_worker |
| lib.rs | graphile_worker |
| version | 0.11.1 |
| created_at | 2024-01-31 17:31:36.752196+00 |
| updated_at | 2025-12-31 09:37:05.861596+00 |
| description | High performance Rust/PostgreSQL job queue (also suitable for getting jobs generated by PostgreSQL triggers/functions out into a different work queue) |
| homepage | https://docs.rs/graphile_worker |
| repository | https://github.com/leo91000/graphile_worker |
| max_upload_size | |
| id | 1121996 |
| size | 675,144 |
A powerful PostgreSQL-backed job queue for Rust applications, based on Graphile Worker. This is a complete Rust rewrite that offers excellent performance, reliability, and a convenient API.
Graphile Worker RS allows you to run jobs (such as sending emails, performing calculations, generating PDFs) in the background, so your HTTP responses and application code remain fast and responsive. It's ideal for any PostgreSQL-backed Rust application.
Key highlights:
SKIP LOCKED for efficient job fetchingLISTEN/NOTIFYThis port is mostly compatible with the original Graphile Worker, meaning you can run it side by side with the Node.js version. The key differences are:
Add the library to your project:
cargo add graphile_worker
A task consists of a struct that implements the TaskHandler trait. Each task has:
Serialize/Deserialize for the payloadrun method that contains the task's logicuse serde::{Deserialize, Serialize};
use graphile_worker::{WorkerContext, TaskHandler, IntoTaskHandlerResult};
#[derive(Deserialize, Serialize)]
struct SendEmail {
to: String,
subject: String,
body: String,
}
impl TaskHandler for SendEmail {
const IDENTIFIER: &'static str = "send_email";
async fn run(self, _ctx: WorkerContext) -> impl IntoTaskHandlerResult {
println!("Sending email to {} with subject '{}'", self.to, self.subject);
// Email sending logic would go here
Ok::<(), String>(())
}
}
Set up the worker with your configuration options and run it:
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
// Create a PostgreSQL connection pool
let pg_pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(5)
.connect("postgres://postgres:password@localhost/mydb")
.await?;
// Initialize and run the worker
graphile_worker::WorkerOptions::default()
.concurrency(5) // Process up to 5 jobs concurrently
.schema("graphile_worker") // Use this PostgreSQL schema
.define_job::<SendEmail>() // Register the task handler
.pg_pool(pg_pool) // Provide the database connection
.init() // Initialize the worker
.await?
.run() // Start processing jobs
.await?;
Ok(())
}
Graphile Worker installs OS-level signal handlers (like SIGINT/SIGTERM) so
it can shut down gracefully when you press Ctrl+C. If your application already
owns the shutdown lifecycle, disable the built-in listeners and call
Worker::request_shutdown() when your orchestrator asks the worker to stop:
let worker = graphile_worker::WorkerOptions::default()
.listen_os_shutdown_signals(false) // prevent installing Ctrl+C handlers
// ... other configuration
.init()
.await?;
tokio::pin! {
let run_loop = worker.run();
}
tokio::select! {
// Main worker loop
result = &mut run_loop => result?,
// Notify the worker when the host framework wants to stop
() = on_shutdown() => {
worker.request_shutdown();
(&mut run_loop).await; // drain gracefully before returning
}
}
Connect to your database and run the following SQL:
SELECT graphile_worker.add_job(
'send_email',
json_build_object(
'to', 'user@example.com',
'subject', 'Welcome to our app!',
'body', 'Thanks for signing up.'
)
);
// Get a WorkerUtils instance to manage jobs
let utils = worker.create_utils();
// Type-safe method (recommended):
utils.add_job(
SendEmail {
to: "user@example.com".to_string(),
subject: "Welcome to our app!".to_string(),
body: "Thanks for signing up.".to_string(),
},
Default::default(), // Use default job options
).await?;
// Or use the raw method when type isn't available:
utils.add_raw_job(
"send_email",
serde_json::json!({
"to": "user@example.com",
"subject": "Welcome to our app!",
"body": "Thanks for signing up."
}),
Default::default(),
).await?;
For efficiency when adding many jobs at once, use batch methods:
// Batch add jobs of the same type (type-safe)
let spec = JobSpec::default();
utils.add_jobs::<SendEmail>(&[
(SendEmail { to: "user1@example.com".into(), subject: "Hello".into(), body: "...".into() }, &spec),
(SendEmail { to: "user2@example.com".into(), subject: "Hello".into(), body: "...".into() }, &spec),
(SendEmail { to: "user3@example.com".into(), subject: "Hello".into(), body: "...".into() }, &spec),
]).await?;
// Batch add jobs of different types (dynamic)
utils.add_raw_jobs(&[
RawJobSpec {
identifier: "send_email".into(),
payload: serde_json::json!({ "to": "user@example.com", "subject": "Hi" }),
spec: JobSpec::default(),
},
RawJobSpec {
identifier: "process_payment".into(),
payload: serde_json::json!({ "user_id": 123, "amount": 50 }),
spec: JobSpec::default(),
},
]).await?;
You can provide shared state to all your tasks using extensions:
use serde::{Deserialize, Serialize};
use graphile_worker::{WorkerContext, TaskHandler, IntoTaskHandlerResult};
use std::sync::{Arc, atomic::{AtomicUsize, Ordering::SeqCst}};
// Define your shared state
#[derive(Clone, Debug)]
struct AppState {
db_client: Arc<DatabaseClient>,
api_key: String,
counter: Arc<AtomicUsize>,
}
// Example database client (just for demonstration)
struct DatabaseClient;
impl DatabaseClient {
fn new() -> Self { Self }
async fn find_user(&self, _user_id: &str) -> Result<(), String> { Ok(()) }
}
#[derive(Deserialize, Serialize)]
struct ProcessUserTask {
user_id: String,
}
impl TaskHandler for ProcessUserTask {
const IDENTIFIER: &'static str = "process_user";
async fn run(self, ctx: WorkerContext) -> impl IntoTaskHandlerResult {
// Access the shared state in your task
let app_state = ctx.get_ext::<AppState>().unwrap();
let count = app_state.counter.fetch_add(1, SeqCst);
// Use shared resources
app_state.db_client.find_user(&self.user_id).await?;
println!("Processed user {}, task count: {}", self.user_id, count);
Ok::<(), String>(())
}
}
// Add the extension when configuring the worker
let app_state = AppState {
db_client: Arc::new(DatabaseClient::new()),
api_key: "secret_key".to_string(),
counter: Arc::new(AtomicUsize::new(0)),
};
graphile_worker::WorkerOptions::default()
.add_extension(app_state)
.define_job::<ProcessUserTask>()
// ... other configuration
.init()
.await?;
You can customize how and when jobs run with the JobSpec builder:
use graphile_worker::{JobSpecBuilder, JobKeyMode};
use chrono::Utc;
// Schedule a job to run after 5 minutes with high priority
let job_spec = JobSpecBuilder::new()
.run_at(Utc::now() + chrono::Duration::minutes(5))
.priority(10)
.job_key("welcome_email_user_123") // Unique identifier for deduplication
.job_key_mode(JobKeyMode::Replace) // Replace existing jobs with this key
.max_attempts(5) // Max retry attempts (default is 25)
.build();
utils.add_job(SendEmail { /* ... */ }, job_spec).await?;
Jobs with the same queue name run in series (one after another) rather than in parallel:
// These jobs will run one after another, not concurrently
let spec1 = JobSpecBuilder::new()
.queue_name("user_123_operations")
.build();
let spec2 = JobSpecBuilder::new()
.queue_name("user_123_operations")
.build();
utils.add_job(UpdateProfile { /* ... */ }, spec1).await?;
utils.add_job(SendEmail { /* ... */ }, spec2).await?;
You can schedule recurring jobs using crontab syntax:
// Run a task daily at 8:00 AM
let worker = WorkerOptions::default()
.with_crontab("0 8 * * * send_daily_report")?
.define_job::<SendDailyReport>()
// ... other configuration
.init()
.await?;
The Local Queue feature batch-fetches jobs from the database and caches them locally, significantly reducing database load in high-throughput scenarios.
use graphile_worker::{WorkerOptions, LocalQueueConfig, RefetchDelayConfig};
use std::time::Duration;
let worker = WorkerOptions::default()
.local_queue(
LocalQueueConfig::default()
.with_size(100) // Cache up to 100 jobs
.with_ttl(Duration::from_secs(300)) // Return unclaimed jobs after 5 minutes
.with_refetch_delay(
RefetchDelayConfig::default()
.with_duration(Duration::from_millis(100)) // Delay between refetches
.with_threshold(10) // Refetch when queue drops below 10
)
)
.define_job::<SendEmail>()
.pg_pool(pg_pool)
.init()
.await?;
The Local Queue operates in several modes:
Key benefits:
You can observe and intercept job lifecycle events using plugins that implement the Plugin trait. This is useful for logging, metrics, validation, and custom job handling logic.
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use graphile_worker::{
Plugin, HookRegistry, HookResult,
WorkerStart, JobStart, JobComplete, JobFail, BeforeJobRun,
};
struct MetricsPlugin {
jobs_started: AtomicU64,
jobs_completed: AtomicU64,
}
impl Plugin for MetricsPlugin {
fn register(self, hooks: &mut HookRegistry) {
hooks.on(WorkerStart, async |ctx| {
println!("Worker {} started", ctx.worker_id);
});
let jobs_started = Arc::new(self.jobs_started);
let jobs_completed = Arc::new(self.jobs_completed);
{
let jobs_started = jobs_started.clone();
hooks.on(JobStart, move |ctx| {
let jobs_started = jobs_started.clone();
async move {
jobs_started.fetch_add(1, Ordering::Relaxed);
println!("Job {} started", ctx.job.id());
}
});
}
{
let jobs_completed = jobs_completed.clone();
hooks.on(JobComplete, move |ctx| {
let jobs_completed = jobs_completed.clone();
async move {
jobs_completed.fetch_add(1, Ordering::Relaxed);
println!("Job {} completed in {:?}", ctx.job.id(), ctx.duration);
}
});
}
hooks.on(JobFail, async |ctx| {
println!("Job {} failed: {}", ctx.job.id(), ctx.error);
});
}
}
The BeforeJobRun and AfterJobRun hooks can intercept jobs and change their behavior:
struct ValidationPlugin;
impl Plugin for ValidationPlugin {
fn register(self, hooks: &mut HookRegistry) {
hooks.on(BeforeJobRun, async |ctx| {
// Skip jobs with a "skip" flag in their payload
if ctx.payload.get("skip").and_then(|v| v.as_bool()).unwrap_or(false) {
return HookResult::Skip;
}
// Fail jobs with invalid data
if ctx.payload.get("invalid").is_some() {
return HookResult::Fail("Invalid payload".into());
}
// Continue with normal execution
HookResult::Continue
});
}
}
Add plugins when configuring the worker:
let worker = WorkerOptions::default()
.define_job::<SendEmail>()
.add_plugin(MetricsPlugin::new())
.add_plugin(ValidationPlugin)
.pg_pool(pg_pool)
.init()
.await?;
Multiple plugins can be registered and they will all receive hook calls in the order they were added.
| Hook | Type | Description |
|---|---|---|
WorkerInit |
Observer | Called when worker is initializing |
WorkerStart |
Observer | Called when worker starts processing |
WorkerShutdown |
Observer | Called when worker is shutting down |
JobFetch |
Observer | Called when a job is fetched from the queue |
JobStart |
Observer | Called before a job starts executing |
JobComplete |
Observer | Called after a job completes successfully |
JobFail |
Observer | Called when a job fails (will retry) |
JobPermanentlyFail |
Observer | Called when a job exceeds max attempts |
CronTick |
Observer | Called on each cron scheduler tick |
CronJobScheduled |
Observer | Called when a cron job is scheduled |
LocalQueueInit |
Observer | Called when local queue is initialized |
LocalQueueSetMode |
Observer | Called when local queue changes mode |
LocalQueueGetJobsComplete |
Observer | Called after batch fetching jobs |
LocalQueueReturnJobs |
Observer | Called when jobs are returned to database |
LocalQueueRefetchDelayStart |
Observer | Called when refetch delay starts |
LocalQueueRefetchDelayAbort |
Observer | Called when refetch delay is aborted |
LocalQueueRefetchDelayExpired |
Observer | Called when refetch delay expires |
BeforeJobRun |
Interceptor | Can skip, fail, or continue job execution |
AfterJobRun |
Interceptor | Can modify the job result after execution |
BeforeJobSchedule |
Interceptor | Can skip, fail, or transform job before scheduling |
The WorkerUtils class provides methods for managing jobs:
// Get a WorkerUtils instance
let utils = worker.create_utils();
// Remove a job by its key
utils.remove_job("job_key_123").await?;
// Mark jobs as completed
utils.complete_jobs(&[job_id1, job_id2]).await?;
// Permanently fail jobs with a reason
utils.permanently_fail_jobs(&[job_id3, job_id4], "Invalid data").await?;
// Reschedule jobs
let options = RescheduleJobOptions {
run_at: Some(Utc::now() + chrono::Duration::minutes(60)),
priority: Some(5),
max_attempts: Some(3),
..Default::default()
};
utils.reschedule_jobs(&[job_id5, job_id6], options).await?;
// Run database cleanup tasks
utils.cleanup(&[
CleanupTask::DeletePermenantlyFailedJobs,
CleanupTask::GcTaskIdentifiers,
CleanupTask::GcJobQueues,
]).await?;
LISTEN/NOTIFY for immediate job notificationsSKIP LOCKED for efficient job fetchingrun_atjob_keygenerated always as (expression) featureProduction ready but the API may continue to evolve. If you encounter any issues or have feature requests, please open an issue on GitHub.
This library is a Rust port of the excellent Graphile Worker by Benjie Gillam. If you find this library useful, please consider sponsoring Benjie's work, as all the research and architecture design was done by him.
MIT License - See LICENSE.md