Laravel Queue System: Background Jobs

QUEUE_CONNECTION=redis

REDIS_HOST=127.0.0.1
REDIS_PASSWORD=null
REDIS_PORT=6379

# Optional: Use a dedicated Redis database for queues
REDIS_QUEUE_DB=1

<?php

return [
    'default' => env('QUEUE_CONNECTION', 'sync'),

    'connections' => [
        'redis' => [
            'driver' => 'redis',
            'connection' => 'default',
            'queue' => env('REDIS_QUEUE', 'default'),
            'retry_after' => 90,  // Seconds before job is retried if not acknowledged
            'block_for' => null,  // Seconds to block while waiting for jobs
            'after_commit' => false,  // Dispatch after DB transaction commits
        ],

        'database' => [
            'driver' => 'database',
            'table' => 'jobs',
            'queue' => 'default',
            'retry_after' => 90,
            'after_commit' => false,
        ],

        'sqs' => [
            'driver' => 'sqs',
            'key' => env('AWS_ACCESS_KEY_ID'),
            'secret' => env('AWS_SECRET_ACCESS_KEY'),
            'prefix' => env('SQS_PREFIX', 'https://sqs.us-east-1.amazonaws.com/your-account-id'),
            'queue' => env('SQS_QUEUE', 'default'),
            'suffix' => env('SQS_SUFFIX'),
            'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
            'after_commit' => false,
        ],
    ],

    // Define which jobs go to which queues
    'failed' => [
        'driver' => env('QUEUE_FAILED_DRIVER', 'database-uuids'),
        'database' => env('DB_CONNECTION', 'mysql'),
        'table' => 'failed_jobs',
    ],
];

# Create a basic job
php artisan make:job SendWelcomeEmail

# Create a job with the ShouldQueue interface
php artisan make:job ProcessPodcastUpload --queued

<?php

namespace App\Jobs;

use App\Models\User;
use App\Mail\WelcomeEmail;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Mail;
use Illuminate\Support\Facades\Log;

class SendWelcomeEmail implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    /**
     * The number of times the job may be attempted.
     */
    public int $tries = 3;

    /**
     * The number of seconds to wait before retrying the job.
     */
    public int $backoff = 30;

    /**
     * The maximum number of unhandled exceptions to allow before failing.
     */
    public int $maxExceptions = 2;

    /**
     * The number of seconds the job can run before timing out.
     */
    public int $timeout = 60;

    /**
     * Create a new job instance.
     */
    public function __construct(
        public User $user,
        public string $referralCode = ''
    ) {}

    /**
     * Execute the job.
     */
    public function handle(): void
    {
        Log::info("Sending welcome email to {$this->user->email}");

        Mail::to($this->user->email)
            ->send(new WelcomeEmail($this->user, $this->referralCode));

        // Update user record
        $this->user->update(['welcome_email_sent_at' => now()]);

        Log::info("Welcome email sent successfully to {$this->user->email}");
    }

    /**
     * Handle a job failure.
     */
    public function failed(\Throwable $exception): void
    {
        Log::error("Failed to send welcome email to {$this->user->email}", [
            'user_id' => $this->user->id,
            'error' => $exception->getMessage(),
        ]);

        // Notify admin of the failure
        // Notification::route('slack', config('services.slack.webhook'))
        //     ->notify(new JobFailedNotification($this, $exception));
    }

    /**
     * Determine the middleware the job should pass through.
     */
    public function middleware(): array
    {
        return [
            new \Illuminate\Queue\Middleware\WithoutOverlapping($this->user->id),
        ];
    }

    /**
     * Get the tags that should be assigned to the job.
     */
    public function tags(): array
    {
        return ['email', 'welcome', 'user:' . $this->user->id];
    }
}

<?php

use App\Jobs\ProcessPodcast;
use App\Jobs\GenerateTranscript;
use App\Jobs\NotifySubscribers;
use Illuminate\Support\Facades\Bus;

// Basic chain
Bus::chain([
    new ProcessPodcast($podcast),
    new GenerateTranscript($podcast),
    new NotifySubscribers($podcast),
])->dispatch();

// Chain with error handling
Bus::chain([
    new ProcessPodcast($podcast),
    new GenerateTranscript($podcast),
    new NotifySubscribers($podcast),
])
->onQueue('podcasts')
->onConnection('redis')
->catch(function (Throwable $e) use ($podcast) {
    // Handle chain failure
    Log::error("Podcast chain failed: {$e->getMessage()}", [
        'podcast_id' => $podcast->id
    ]);
    
    $podcast->update(['status' => 'failed']);
})
->dispatch();

// Dynamic chain based on conditions
$jobs = [new ProcessPodcast($podcast)];

if ($podcast->wants_transcript) {
    $jobs[] = new GenerateTranscript($podcast);
}

if ($podcast->notify_subscribers) {
    $jobs[] = new NotifySubscribers($podcast);
}

Bus::chain($jobs)->dispatch();

<?php

use App\Jobs\ProcessImage;
use App\Models\Gallery;
use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;

class GalleryController extends Controller
{
    public function processGallery(Gallery $gallery)
    {
        $jobs = $gallery->images->map(function ($image) {
            return new ProcessImage($image);
        })->toArray();

        $batch = Bus::batch($jobs)
            ->then(function (Batch $batch) use ($gallery) {
                // All jobs completed successfully
                $gallery->update([
                    'status' => 'processed',
                    'processed_at' => now(),
                ]);
                
                // Notify user
                $gallery->user->notify(new GalleryProcessed($gallery));
            })
            ->catch(function (Batch $batch, Throwable $e) use ($gallery) {
                // First batch job failure detected
                Log::error("Gallery processing failed", [
                    'gallery_id' => $gallery->id,
                    'error' => $e->getMessage(),
                    'failed_jobs' => $batch->failedJobs,
                ]);
            })
            ->finally(function (Batch $batch) use ($gallery) {
                // Batch finished (success or failure)
                $gallery->update([
                    'batch_id' => null,  // Clear batch reference
                    'jobs_processed' => $batch->totalJobs - $batch->pendingJobs,
                    'jobs_failed' => $batch->failedJobs,
                ]);
            })
            ->name("Process Gallery #{$gallery->id}")
            ->onQueue('image-processing')
            ->allowFailures()  // Continue processing even if some jobs fail
            ->dispatch();

        // Store batch ID for progress tracking
        $gallery->update(['batch_id' => $batch->id]);

        return response()->json([
            'message' => 'Gallery processing started',
            'batch_id' => $batch->id,
        ]);
    }

    public function checkProgress(Gallery $gallery)
    {
        $batch = Bus::findBatch($gallery->batch_id);

        if (!$batch) {
            return response()->json(['error' => 'Batch not found'], 404);
        }

        return response()->json([
            'id' => $batch->id,
            'name' => $batch->name,
            'total_jobs' => $batch->totalJobs,
            'pending_jobs' => $batch->pendingJobs,
            'failed_jobs' => $batch->failedJobs,
            'progress' => $batch->progress(),
            'finished' => $batch->finished(),
            'cancelled' => $batch->cancelled(),
        ]);
    }
}

<?php

namespace App\Jobs;

use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\Middleware\RateLimited;
use Illuminate\Queue\Middleware\ThrottlesExceptions;
use Illuminate\Queue\Middleware\WithoutOverlapping;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\RateLimiter;

class SyncToExternalAPI implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public function __construct(
        public Model $record
    ) {}

    /**
     * Get the middleware the job should pass through.
     */
    public function middleware(): array
    {
        return [
            // Global rate limit: max 60 jobs per minute across all workers
            new RateLimited('external-api'),
            
            // Throttle exceptions: if job fails, wait before retrying
            (new ThrottlesExceptions(3, 5))  // 3 attempts, 5 min cooldown
                ->backoff(5),  // Wait 5 minutes after throttling
            
            // Prevent overlapping: only one job per record at a time
            (new WithoutOverlapping($this->record->id))
                ->releaseAfter(60)  // Release lock after 60 seconds
                ->expireAfter(180), // Lock expires after 3 minutes
        ];
    }

    public function handle(): void
    {
        // Your API sync logic here
        $this->syncToApi();
    }

    /**
     * Determine the time at which the job should timeout.
     */
    public function retryUntil(): DateTime
    {
        return now()->addHours(24);  // Keep retrying for 24 hours
    }
}

// Define the rate limiter in AppServiceProvider
class AppServiceProvider extends ServiceProvider
{
    public function boot(): void
    {
        RateLimiter::for('external-api', function (object $job) {
            return Limit::perMinute(60);  // 60 jobs per minute
        });

        // Or with more sophisticated limits
        RateLimiter::for('stripe-api', function (object $job) {
            return [
                Limit::perSecond(10),      // 10 per second
                Limit::perMinute(100),     // 100 per minute  
                Limit::perHour(1000),      // 1000 per hour
            ];
        });

        // User-specific rate limiting
        RateLimiter::for('user-exports', function (object $job) {
            return Limit::perHour(5)->by($job->user->id);
        });
    }
}

<?php

return [
    'domain' => env('HORIZON_DOMAIN'),
    'path' => 'horizon',
    
    // Redis connection to use
    'use' => 'default',
    
    // Prefix for Horizon keys in Redis
    'prefix' => env('HORIZON_PREFIX', 'horizon:'),
    
    // Dashboard route middleware
    'middleware' => ['web', 'auth', 'can:viewHorizon'],
    
    // Queue worker configuration
    'defaults' => [
        'supervisor-1' => [
            'connection' => 'redis',
            'queue' => ['default'],
            'balance' => 'auto',      // auto, simple, or false
            'autoScalingStrategy' => 'time',  // time or size
            'maxProcesses' => 10,
            'maxTime' => 0,
            'maxJobs' => 0,
            'memory' => 128,
            'tries' => 3,
            'timeout' => 60,
            'nice' => 0,
        ],
    ],
    
    'environments' => [
        'production' => [
            // High priority jobs - always process immediately
            'supervisor-high' => [
                'connection' => 'redis',
                'queue' => ['high', 'notifications'],
                'balance' => 'false',  // Fixed number of workers
                'processes' => 5,
                'tries' => 3,
                'timeout' => 30,
            ],
            
            // Default queue with auto-scaling
            'supervisor-default' => [
                'connection' => 'redis',
                'queue' => ['default'],
                'balance' => 'auto',
                'minProcesses' => 1,
                'maxProcesses' => 10,
                'balanceMaxShift' => 1,  // Max processes to add/remove at once
                'balanceCooldown' => 3,  // Seconds between scaling decisions
            ],
            
            // Low priority, long-running jobs
            'supervisor-low' => [
                'connection' => 'redis',
                'queue' => ['low', 'reports', 'exports'],
                'balance' => 'auto',
                'minProcesses' => 1,
                'maxProcesses' => 3,
                'tries' => 1,
                'timeout' => 1800,  // 30 minutes
            ],
        ],
        
        'local' => [
            'supervisor-1' => [
                'connection' => 'redis',
                'queue' => ['high', 'default', 'low'],
                'balance' => 'false',
                'processes' => 3,
            ],
        ],
    ],
];

[program:horizon]
process_name=%(program_name)s
command=php /var/www/app/artisan horizon
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=www-data
numprocs=1
redirect_stderr=true
stdout_logfile=/var/www/app/storage/logs/horizon.log
stopwaitsecs=3600
stopsignal=SIGTERM

; For queue:work without Horizon
[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/app/artisan queue:work redis --sleep=3 --tries=3 --max-time=3600
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=www-data
numprocs=8
redirect_stderr=true
stdout_logfile=/var/www/app/storage/logs/worker.log
stopwaitsecs=3600

#!/bin/bash

# Put application in maintenance mode
php artisan down

# Pull latest code
git pull origin main

# Install dependencies
composer install --no-dev --optimize-autoloader

# Run migrations
php artisan migrate --force

# Clear and rebuild caches
php artisan config:cache
php artisan route:cache
php artisan view:cache

# Restart Horizon gracefully (finish current jobs, then restart)
php artisan horizon:terminate

# Or for queue:work, restart via Supervisor
# supervisorctl restart laravel-worker:*

# Bring application back online
php artisan up

echo "Deployment complete!"