Middleware

Middleware wraps job execution in an onion-style pipeline. Each middleware can run logic before and after the job, short-circuit execution, or catch exceptions. Queuety ships five built-in middleware and supports custom middleware through the Contracts\Middleware interface.

Middleware is available for dispatchable job classes (classes implementing Contracts\Job with the Dispatchable trait).

The Contracts\Middleware interface

use Queuety\Contracts\Middleware;

interface Middleware {
    public function handle( object $job, \Closure $next ): void;
}

The $job parameter is the job instance being processed. Call $next( $job ) to pass execution to the next middleware in the pipeline (or to the job's handle() method if this is the last middleware).

Adding middleware to a job

Define a middleware() method on your job class that returns an array of middleware instances:

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;

class SyncInventoryJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $sku,
    ) {}

    public function handle(): void {
        // Sync inventory for this SKU...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 30 ),
            new RateLimited( max: 100, window: 60 ),
        ];
    }
}

Pipeline execution order

Middleware are applied in array order, with the first middleware being the outermost layer. If your middleware() method returns [A, B, C], the execution order is:

A::handle() -> B::handle() -> C::handle() -> $job->handle()

Each middleware wraps the next. A runs first before the job and last after the job, making it ideal for timing or try/catch wrappers.

Writing custom middleware

Implement the Contracts\Middleware interface:

use Queuety\Contracts\Middleware;

class LogExecution implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        $class = get_class( $job );
        error_log( "[Queuety] Starting: {$class}" );

        $start = microtime( true );

        try {
            $next( $job );
        } finally {
            $elapsed = round( microtime( true ) - $start, 3 );
            error_log( "[Queuety] Finished: {$class} in {$elapsed}s" );
        }
    }
}

Use it on a job:

public function middleware(): array {
    return [
        new LogExecution(),
    ];
}

Short-circuiting

A middleware can skip the job entirely by not calling $next():

class SkipOnMaintenanceMode implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        if ( wp_is_maintenance_mode() ) {
            return; // Job silently skipped.
        }

        $next( $job );
    }
}

Built-in middleware

ThrottlesExceptions

Throttles exceptions from a job handler to prevent job storms. If a handler has thrown too many exceptions within a time window, subsequent jobs for that handler are rejected with a RuntimeException instead of being processed. This prevents a failing external service from burning through all retry attempts across many jobs at once.

use Queuety\Middleware\ThrottlesExceptions;

public function middleware(): array {
    return [
        new ThrottlesExceptions( max_attempts: 10, decay_minutes: 10 ),
    ];
}

Exception counts are tracked per handler class in Queuety's locks table and shared across all worker processes. When the threshold is reached, the middleware throws a RuntimeException without calling the job, and the worker treats that like any other failure: the attempt counts towards max_attempts, the job is retried with the normal backoff, and it is buried once its attempts run out.

Example: if an external API is down, the first 10 failures are processed normally (each incrementing the exception counter). Further attempts fail fast without calling the API until 10 minutes pass without a new exception; the window restarts with every recorded exception.

RateLimited

Enforces a maximum number of executions within a time window. If the limit is exceeded inside the middleware, a RateLimitExceededException is thrown; the worker handles it as an ordinary failed attempt, retried with the normal backoff and counted towards max_attempts. Once the middleware has registered the limit with the worker's rate limiter, later jobs for the handler are also put back to pending before they run while the limit is reached (see Rate Limiting).

use Queuety\Middleware\RateLimited;

public function middleware(): array {
    return [
        new RateLimited( max: 60, window: 60 ), // 60 per minute
    ];
}

The rate limiter uses Queuety's internal RateLimiter instance, so limits are enforced across all worker processes.

Timeout

Arms a per-job execution timeout with pcntl_alarm and is meant to throw a TimeoutException when the job runs too long. The timeout is not enforced today (#10): Queuety does not enable async signals, so the alarm handler does not run while the job is executing. The middleware also resets the alarm when it returns, which cancels the worker's own job timeout alarm.

use Queuety\Middleware\Timeout;

public function middleware(): array {
    return [
        new Timeout( seconds: 30 ),
    ];
}

Requires the pcntl PHP extension. If pcntl is not available, the middleware falls through silently (no timeout is enforced).

UniqueJob

Prevents concurrent execution of jobs with the same key. Uses a database lock table to ensure only one instance runs at a time. If a lock is already held, the job is silently skipped (not retried).

use Queuety\Middleware\UniqueJob;

public function middleware(): array {
    return [
        new UniqueJob(),                         // Lock key defaults to the job class name
        new UniqueJob( key: 'sync_inventory' ),   // Custom lock key
    ];
}

WithoutOverlapping

Prevents overlapping execution of jobs sharing a logical key, with an automatic lock expiry to prevent dead locks from crashed workers. Similar to UniqueJob but designed for long-running operations where the lock should auto-release after a timeout.

use Queuety\Middleware\WithoutOverlapping;

public function middleware(): array {
    return [
        new WithoutOverlapping( key: 'import_products', release_after: 600 ),
    ];
}

If a lock is already held, the job is silently skipped. Expired locks are cleaned up automatically before each acquisition attempt.

Example: combining middleware

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;
use Queuety\Middleware\WithoutOverlapping;

class ImportProductsJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly int $supplier_id,
    ) {}

    public function handle(): void {
        // Fetch and import products from the supplier...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 300 ),
            new WithoutOverlapping(
                key: "import_supplier_{$this->supplier_id}",
                release_after: 600,
            ),
            new RateLimited( max: 5, window: 60 ),
        ];
    }
}

This job will:

1. Be killed if it runs longer than 5 minutes (Timeout)
2. Be skipped if another import for the same supplier is already running (WithoutOverlapping)
3. Be rate limited to 5 imports per minute across all suppliers (RateLimited)