Struct WorkerOptions 

pub struct WorkerOptions { /* private fields */ }

Configuration options for initializing a Graphile Worker instance.

WorkerOptions provides a builder-style API for configuring a worker instance, including database connection settings, concurrency, job definitions, and more.

Example

use graphile_worker::WorkerOptions;
use graphile_worker::{TaskHandler, WorkerContext, IntoTaskHandlerResult};
use std::time::Duration;
use serde::{Deserialize, Serialize};

#[derive(Deserialize, Serialize)]
struct MyTask { data: String }

impl TaskHandler for MyTask {
    const IDENTIFIER: &'static str = "my_task";
    async fn run(self, _ctx: WorkerContext) -> impl IntoTaskHandlerResult {
        Ok::<(), String>(())
    }
}

async fn example() -> Result<(), Box<dyn std::error::Error>> {
    let worker = WorkerOptions::default()
        .concurrency(5)
        .schema("my_app_worker")
        .poll_interval(Duration::from_millis(500))
        .database_url("postgres://user:password@localhost/mydb")
        .define_job::<MyTask>()
        .init()
        .await?;
     
    // Start the worker
    worker.run().await?;
     
    Ok(())
}

Implementations

impl WorkerOptions

pub fn complete_job_batch_delay(self, delay: Duration) -> Self

Sets the delay before flushing batched job completions.

When configured, job completions are collected in a batch and flushed together after the specified delay. This reduces the number of SQL round trips and can significantly improve throughput.

Arguments

• delay - The duration to wait before flushing the batch. A small value like 1-5ms is recommended.

Example

let worker = WorkerOptions::default()
    .complete_job_batch_delay(Duration::from_millis(5));

pub fn fail_job_batch_delay(self, delay: Duration) -> Self

Sets the delay before flushing batched job failures.

When configured, permanent job failures are collected and flushed together after the specified delay. Retryable failures are still processed individually to ensure proper backoff timing.

Arguments

• delay - The duration to wait before flushing the batch. A small value like 1-5ms is recommended.

Example

let worker = WorkerOptions::default()
    .fail_job_batch_delay(Duration::from_millis(5));

impl WorkerOptions

pub fn schema(self, value: impl Into<Schema>) -> Self

Sets the PostgreSQL schema name for Graphile Worker tables.

Isolates Graphile Worker tables in a separate schema, which keeps the database organized or allows running multiple independent worker instances in the same database.

Arguments

• value - The schema name to use

Default

If not specified, the schema name defaults to “graphile_worker”.

pub fn concurrency(self, value: usize) -> Self

Sets the number of jobs that can be processed concurrently.

Controls how many jobs the worker processes simultaneously. Setting an appropriate concurrency level depends on workload:

• CPU-intensive tasks: Set close to the number of cores
• I/O-intensive tasks: Can often use higher values (10-20+)

Arguments

• value - The maximum number of concurrent jobs

Default

If not specified, defaults to the number of logical CPUs in the system.

Panics

Panics if the value is 0, as at least one job must be processable.

pub fn poll_interval(self, value: Duration) -> Self

Sets how often the worker checks the database for new jobs.

Controls the polling interval for checking for new jobs when PostgreSQL notification delivery fails or for jobs scheduled in the future.

Arguments

• value - The interval between database polls

Default

If not specified, defaults to 1000 milliseconds (1 second).

Note

Lower values increase responsiveness but may increase database load. For most applications, the default value is appropriate.

pub fn use_notification_delivery(self, value: bool) -> Self

Sets whether to use PostgreSQL notification delivery.

Arguments

• value - True to use notification delivery, false to use polling only

Default

If not specified, defaults to true

pub fn use_local_time(self, value: bool) -> Self

Sets whether to use local application time or database time for timestamps.

When use_local_time is true, the application’s Utc::now() is used for timestamps, which can help handle clock drift between the application server and database server. When false (default), PostgreSQL’s now() is used instead.

Affects job fetching, job scheduling, and crontab scheduling.

Arguments

• value - True to use application time, false to use database time

Default

If not specified, defaults to false (use PostgreSQL server time).

Note

Using PostgreSQL server time is recommended for consistent behavior across multiple worker instances, especially in distributed deployments.

impl WorkerOptions

pub fn with_cron<C: CronInput>(self, cron: C) -> C::Output

Adds cron entries for scheduled jobs.

This accepts typed cron builders, raw Crontab values, and crontab text. Typed inputs return WorkerOptions directly; text input returns Result<WorkerOptions, CrontabParseError>.

Typed example

let options = WorkerOptions::default()
    .define_job::<SendDigest>()
    .with_cron(
        Cron::daily_at::<SendDigest>(8, 0)
            .expect("valid cron schedule")
            .fill(CrontabFill::hours(1)),
    );

Crontab text example

let options = WorkerOptions::default()
    .with_cron("0 8 * * * send_digest")?;

pub fn with_crons<I, C>(self, crontabs: I) -> Self

where I: IntoIterator<Item = C>, C: Into<Crontab>,

Adds typed cron entries for scheduled jobs.

pub fn with_crontab(self, input: &str) -> Result<Self, CrontabParseError>

👎Deprecated:

use WorkerOptions::with_cron(…) instead

Adds crontab text entries for scheduled jobs.

Use Self::with_cron with a string instead.

Arguments

• input - A string containing crontab entries

Returns

• Result<Self, CrontabParseError> - The modified WorkerOptions instance or a parse error

Example

// Run the "send_digest" job at 8:00 AM every day.
let options = WorkerOptions::default()
    .with_cron("0 8 * * * send_digest")?;

impl WorkerOptions

pub fn database(self, value: impl Into<Database>) -> Self

Sets an existing PostgreSQL database connection for the worker to use.

Allows reusing an existing driver-specific pool or configuring the database connection with custom settings before passing it to the worker.

Arguments

• value - The PostgreSQL database connection

Note

If both database and database_url are provided, database takes precedence.

pub fn pg_pool(self, value: PgPool) -> Self

Sets an existing SQLx PostgreSQL pool for the worker to use.

This is a SQLx-only convenience wrapper over Self::database. Prefer Self::database when passing a driver-agnostic database connection.

Note

If both pg_pool and database_url are provided, pg_pool takes precedence.

pub fn database_url(self, value: &str) -> Self

Sets the PostgreSQL database connection URL.

The URL is used to establish a connection to the database if no connection pool is provided.

Arguments

• value - The PostgreSQL connection URL (e.g., “postgres://user:password@localhost/mydb”)

Note

Either database, pg_pool, or database_url must be provided before calling init().

pub fn max_pg_conn(self, value: u32) -> Self

Sets the maximum number of database connections in the pool.

Only applies when creating a new connection pool from a database URL. Ignored if an existing pool is provided.

Arguments

• value - The maximum number of connections

Default

If not specified, defaults to 20 connections.

impl WorkerOptions

pub fn add_extension<T: Clone + Send + Sync + Debug + 'static>( self, value: T, ) -> Self

Adds a custom extension to the worker context.

Extensions provide custom application state or dependencies that are accessible from task handlers. Useful for sharing resources like API clients, configuration, or other state between jobs.

Type Parameters

• T - Type of the extension (must be Clone, Send, Sync, Debug, and ’static)

Arguments

• value - The extension instance to add

Example

let config = AppConfig {
    api_key: "secret".to_string(),
};

let db = Database {
    // Initialize database
};

let options = WorkerOptions::default()
    .add_extension(config)
    .add_extension(db);

pub fn on<E, F, Fut>(self, event: E, handler: F) -> Self

where E: Event, F: Fn(E::Context) -> Fut + Send + Sync + Clone + 'static, Fut: Future<Output = E::Output> + Send + 'static,

Registers an event handler for a specific lifecycle event.

This method allows registering individual handlers for specific events without creating a full plugin. The handler is a closure that receives the event context and returns the appropriate output type.

Type Parameters

• E - The event type to handle

Arguments

• event - The event marker (e.g., JobStart, BeforeJobRun)
• handler - The async handler closure

Example

use graphile_worker::{WorkerOptions, JobStart, BeforeJobRun, HookResult};

let options = WorkerOptions::default()
    .on(JobStart, |ctx| async move {
        println!("Job {} starting", ctx.job.id());
    })
    .on(BeforeJobRun, |ctx| async move {
        if ctx.payload.get("skip").and_then(|v| v.as_bool()).unwrap_or(false) {
            return HookResult::Skip;
        }
        HookResult::Continue
    });

pub fn add_plugin<P: Plugin>(self, plugin: P) -> Self

Adds a lifecycle hook plugin to the worker.

Plugins can observe and intercept various worker events such as job execution, worker startup/shutdown, and cron scheduling. Multiple plugins can be registered and they execute in registration order.

Type Parameters

• P - A type implementing the Plugin trait

Arguments

• plugin - The plugin instance to add

Example

use graphile_worker::{WorkerOptions, Plugin, HookRegistry, JobStart};

struct LoggingPlugin;

impl Plugin for LoggingPlugin {
    fn register(self, hooks: &mut HookRegistry) {
        hooks.on::<JobStart>(|ctx| async move {
            println!("Job {} starting", ctx.job.id());
        });
    }
}

let options = WorkerOptions::default()
    .add_plugin(LoggingPlugin);

impl WorkerOptions

pub async fn init(self) -> Result<Worker, WorkerBuildError>

Initializes a worker with the configured options.

Process:

1. Establishes a database connection (using the provided pool or creating one from the URL)
2. Runs database migrations to ensure the schema is up to date
3. Registers the task handlers
4. Initializes the worker with a random ID and the configured settings

Returns

• Result<Worker, WorkerBuildError> - A fully configured worker instance or an error

Errors

Can fail if:

• Database connection fails
• Database URL is missing and no pool was provided
• Migrations fail
• Task registration fails

Example

let worker = WorkerOptions::default()
    .database_url("postgres://user:password@localhost/mydb")
    .schema("my_app_worker")
    .define_job::<MyTask>()
    .init()
    .await?;

impl WorkerOptions

pub fn define_job<T: TaskHandler>(self) -> Self

Registers a task handler type with the worker.

Primary way to define what types of jobs this worker can process. Each task handler implements the TaskHandler trait, which defines how to deserialize and run jobs of a specific type.

Type Parameters

• T - A type implementing the TaskHandler trait

Example

let options = WorkerOptions::default()
    .define_job::<SendEmail>();

pub fn define_batch_job<T: BatchTaskHandler>(self) -> Self

Registers a batch task handler type with the worker.

Batch handlers receive a vector of item payloads from a single JSON-array job payload and may return per-item results so successful items are not retried after partial failure.

pub fn define_jobs<I>(self, jobs: I) -> Self

where I: IntoIterator<Item = JobDefinition>,

Registers task handler definitions with the worker.

This is useful when a module or crate exposes the jobs it can process as reusable values.

Arguments

• jobs - Job definitions created with TaskHandler::definition or JobDefinition::of

Example

fn jobs() -> [JobDefinition; 1] {
    [SendEmail::definition()]
}

let options = WorkerOptions::default()
    .define_jobs(jobs());

pub fn add_forbidden_flag(self, flag: &str) -> Self

Adds a flag to the list of forbidden flags.

Jobs with any forbidden flag will be skipped by this worker instance. Can be used to implement specialized workers that only handle certain types of jobs.

Arguments

• flag - The flag to forbid

Example

// This worker will skip jobs with the "high_memory" flag
let options = WorkerOptions::default()
    .add_forbidden_flag("high_memory");

impl WorkerOptions

pub fn local_queue(self, config: LocalQueueConfig) -> Self

Enables the LocalQueue with the specified configuration.

LocalQueue batch-fetches jobs from the database to reduce DB load, trading latency for throughput. Jobs are cached locally and distributed to workers without additional database queries until the cache is empty.

Arguments

• config - The LocalQueue configuration (size and TTL)

Note

When LocalQueue is enabled, jobs may experience slightly higher latency as they wait in the local cache. The cache has a TTL after which unclaimed jobs are returned to the database.

LocalQueueConfig::queue_count can be raised above 1 to run multiple independent local queues in the same worker. size is applied per queue, so total local capacity is size * queue_count. There is no universal best setting for throughput; benchmark realistic jobs with your own PostgreSQL latency, pool size, worker concurrency, and local queue settings before tuning this in production.

Workers with forbidden_flags will bypass the LocalQueue and fetch jobs directly from the database.

Example

let options = WorkerOptions::default()
    .local_queue(
        LocalQueueConfig::default()
            .with_size(100)
            .with_queue_count(2)
            .with_ttl(Duration::from_secs(300))
            .with_refetch_delay(
                RefetchDelayConfig::default()
                    .with_duration(Duration::from_millis(100))
                    .with_threshold(10)
                    .with_max_abort_threshold(500),
            ),
    );

impl WorkerOptions

pub fn worker_recovery(self, config: WorkerRecoveryConfig) -> Self

Configures dead worker recovery.

See crate::WorkerRecoveryConfig for Pro-aligned options such as heartbeat_interval (heartbeatInterval), sweep_interval (sweepInterval), and sweep_threshold (sweepThreshold).

pub fn heartbeat_interval(self, interval: Duration) -> Self

Sets the worker heartbeat interval (heartbeatInterval).

pub fn sweep_interval(self, interval: Duration) -> Self

Sets the inactive worker sweep interval (sweepInterval).

pub fn sweep_threshold(self, threshold: Duration) -> Self

Sets the inactive worker threshold (sweepThreshold).

pub fn recovery_delay(self, delay: Duration) -> Self

Sets the delay before jobs recovered from dead workers become eligible again.

impl WorkerOptions

pub fn worker_shutdown(self, config: WorkerShutdownConfig) -> Self

Configures worker shutdown behavior.

pub fn listen_os_shutdown_signals(self, value: bool) -> Self

Controls whether the worker installs OS-level shutdown signal handlers.

By default Graphile Worker listens to signals like SIGINT/SIGTERM to trigger a graceful shutdown. Embedding applications that already manage signal handling can disable this behavior by setting the value to false.

Arguments

• value - true to install the default OS signal listeners, false to skip them

pub fn shutdown_signal( self, signal: impl Future<Output = ()> + Send + 'static, ) -> Self

Sets a custom application shutdown signal.

The future should complete when the host application requests shutdown. The worker still owns graceful draining after the signal is received.

pub fn shutdown_grace_period(self, period: Duration) -> Self

Sets how long in-flight jobs may continue after a shutdown signal.

pub fn shutdown_interrupted_job_retry_delay(self, delay: Duration) -> Self

Sets the delay before shutdown-aborted jobs are retried.

Trait Implementations

impl Default for WorkerOptions

fn default() -> Self

Returns the “default value” for a type.