src/Illuminate/Queue/CallQueuedHandler.php - 13.x - githubmirror/framework

src/Illuminate/Queue/CallQueuedHandler.php
477 строк14 KB
Taylor Otwell
Attempt at lock fix (#60906)
03 авг 2026, 01:16
03 авг 2026, 01:16f7bb662
<?php

namespace Illuminate\Queue;

use Exception;
use Illuminate\Bus\Batchable;
use Illuminate\Bus\BatchRepository;
use Illuminate\Bus\DebounceLock;
use Illuminate\Bus\Queueable;
use Illuminate\Bus\UniqueLock;
use Illuminate\Contracts\Bus\Dispatcher;
use Illuminate\Contracts\Cache\Factory as CacheFactory;
use Illuminate\Contracts\Cache\LockProvider;
use Illuminate\Contracts\Cache\Repository as Cache;
use Illuminate\Contracts\Container\Container;
use Illuminate\Contracts\Encryption\Encrypter;
use Illuminate\Contracts\Queue\Job;
use Illuminate\Contracts\Queue\ShouldBeUnique;
use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing;
use Illuminate\Database\Eloquent\ModelNotFoundException;
use Illuminate\Events\CallQueuedListener;
use Illuminate\Log\Context\Repository as ContextRepository;
use Illuminate\Pipeline\Pipeline;
use Illuminate\Queue\Events\JobDebounced;
use RuntimeException;

class CallQueuedHandler
{
    /**
     * The bus dispatcher implementation.
     *
     * @var \Illuminate\Contracts\Bus\Dispatcher
     */
    protected $dispatcher;

    /**
     * The container instance.
     *
     * @var \Illuminate\Contracts\Container\Container
     */
    protected $container;

    /**
     * The command currently being processed.
     *
     * @var mixed
     */
    protected $runningCommand;

    /**
     * Create a new handler instance.
     *
     * @param  \Illuminate\Contracts\Bus\Dispatcher  $dispatcher
     * @param  \Illuminate\Contracts\Container\Container  $container
     */
    public function __construct(Dispatcher $dispatcher, Container $container)
    {
        $this->container = $container;
        $this->dispatcher = $dispatcher;
    }

    /**
     * Handle the queued job.
     *
     * @param  \Illuminate\Contracts\Queue\Job  $job
     * @param  array  $data
     * @return void
     */
    public function call(Job $job, array $data)
    {
        try {
            $command = $this->setJobInstanceIfNecessary(
                $job, $this->getCommand($data)
            );
        } catch (ModelNotFoundException $e) {
            return $this->handleModelNotFound($job, $e);
        }

        if ($this->commandShouldBeDebounced($command)) {
            return $this->deleteDebouncedJob($job, $command);
        }

        $this->runningCommand = $command;

        try {
            $this->dispatchThroughMiddleware($job, $command);
        } finally {
            $this->runningCommand = null;
        }

        if (! $job->isReleased() && ! $this->commandShouldBeUniqueUntilProcessing($command)) {
            $this->ensureUniqueJobLockIsReleased($command);
        }

        if (! $job->hasFailed() && ! $job->isReleased()) {
            $this->ensureNextJobInChainIsDispatched($command);
            $this->ensureSuccessfulBatchJobIsRecorded($command);
        }

        if (! $job->isDeletedOrReleased()) {
            $job->delete();
        }
    }

    /**
     * Get the command from the given payload.
     *
     * @param  array  $data
     * @return mixed
     *
     * @throws \RuntimeException
     */
    protected function getCommand(array $data)
    {
        if (str_starts_with($data['command'], 'O:')) {
            return unserialize($data['command']);
        }

        if ($this->container->bound(Encrypter::class)) {
            return unserialize($this->container[Encrypter::class]->decrypt($data['command']));
        }

        throw new RuntimeException('Unable to extract job payload.');
    }

    /**
     * Dispatch the given job / command through its specified middleware.
     *
     * @param  \Illuminate\Contracts\Queue\Job  $job
     * @param  mixed  $command
     * @return mixed
     */
    protected function dispatchThroughMiddleware(Job $job, $command)
    {
        if ($command instanceof \__PHP_Incomplete_Class) {
            throw new Exception('Job is incomplete class: '.json_encode($command));
        }

        $lockReleased = false;

        return (new Pipeline($this->container))->send($command)
            ->through(array_merge(method_exists($command, 'middleware') ? $command->middleware() : [], $command->middleware ?? []))
            ->finally(function ($command) use (&$lockReleased) {
                if (! $lockReleased && $this->commandShouldBeUniqueUntilProcessing($command) && ! $command->job->isReleased() && $this->uniqueJobLockShouldBeReleased($command->job, $command)) {
                    $this->ensureUniqueJobLockIsReleased($command);
                }
            })
            ->then(function ($command) use ($job, &$lockReleased) {
                if ($this->commandShouldBeUniqueUntilProcessing($command) && $this->uniqueJobLockShouldBeReleased($job, $command)) {
                    $this->ensureUniqueJobLockIsReleased($command);

                    $lockReleased = true;
                }

                return $this->dispatcher->dispatchNow(
                    $command, $this->resolveHandler($job, $command)
                );
            });
    }

    /**
     * Resolve the handler for the given command.
     *
     * @param  \Illuminate\Contracts\Queue\Job  $job
     * @param  mixed  $command
     * @return mixed
     */
    protected function resolveHandler($job, $command)
    {
        $handler = $this->dispatcher->getCommandHandler($command) ?: null;

        if ($handler) {
            $this->setJobInstanceIfNecessary($job, $handler);
        }

        return $handler;
    }

    /**
     * Set the job instance of the given class if necessary.
     *
     * @param  \Illuminate\Contracts\Queue\Job  $job
     * @param  mixed  $instance
     * @return mixed
     */
    protected function setJobInstanceIfNecessary(Job $job, $instance)
    {
        if (isset(class_uses_recursive($instance)[InteractsWithQueue::class])) {
            $instance->setJob($job);
        }

        return $instance;
    }

    /**
     * Ensure the next job in the chain is dispatched if applicable.
     *
     * @param  mixed  $command
     * @return void
     */
    protected function ensureNextJobInChainIsDispatched($command)
    {
        if (method_exists($command, 'dispatchNextJobInChain')) {
            $command->dispatchNextJobInChain();
        }
    }

    /**
     * Ensure the batch is notified of the successful job completion.
     *
     * @param  mixed  $command
     * @return void
     */
    protected function ensureSuccessfulBatchJobIsRecorded($command)
    {
        $uses = class_uses_recursive($command);

        if (! isset($uses[Batchable::class], $uses[InteractsWithQueue::class])) {
            return;
        }

        if ($batch = $command->batch()) {
            $batch->recordSuccessfulJob($command->job->uuid());
        }
    }

    /**
     * Determine if the unique job lock can be safely released.
     *
     * @param  \Illuminate\Contracts\Queue\Job  $job
     * @param  mixed  $command
     * @return bool
     */
    protected function uniqueJobLockShouldBeReleased(Job $job, $command)
    {
        return $job->attempts() <= 1 ||
            (isset(class_uses_recursive($command)[Queueable::class]) &&
             is_string($command->uniqueLockOwner ?? null) &&
             $command->uniqueLockOwner !== '');
    }

    /**
     * Ensure the lock for a unique job is released.
     *
     * @param  mixed  $command
     * @return void
     */
    protected function ensureUniqueJobLockIsReleased($command)
    {
        if ($this->commandShouldBeUnique($command)) {
            (new UniqueLock($this->container->make(Cache::class)))->release($command);
        }
    }

    /**
     * Determine if the debounced command was superseded by a newer dispatch.
     *
     * @param  mixed  $command
     * @return bool
     */
    protected function commandShouldBeDebounced($command)
    {
        $owner = $command->debounceOwner ?? '';

        if (empty($owner)) {
            return false;
        }

        $lock = new DebounceLock($this->container->make(Cache::class));

        $currentOwner = $lock->getCurrentOwner($command);

        // Fail-open: if the lock no longer exists (cache eviction, TTL expiry), let the job execute...
        if (is_null($currentOwner)) {
            return false;
        }

        return $currentOwner !== $owner;
    }

    /**
     * Handle a debounced (superseded) job by firing an event and deleting it.
     *
     * @param  \Illuminate\Contracts\Queue\Job  $job
     * @param  mixed  $command
     * @return void
     */
    protected function deleteDebouncedJob($job, $command)
    {
        if ($this->container->bound('events')) {
            $this->container->make('events')->dispatch(
                new JobDebounced($job->getConnectionName(), $job, $command)
            );
        }

        $job->delete();
    }

    /**
     * Determine if the given command should be unique.
     */
    protected function commandShouldBeUnique(mixed $command): bool
    {
        return $command instanceof ShouldBeUnique ||
            ($command instanceof CallQueuedListener && $command->shouldBeUnique());
    }

    /**
     * Determine if the given command should be unique until processing begins.
     */
    protected function commandShouldBeUniqueUntilProcessing(mixed $command): bool
    {
        return $command instanceof ShouldBeUniqueUntilProcessing ||
            ($command instanceof CallQueuedListener && $command->shouldBeUniqueUntilProcessing());
    }

    /**
     * Handle a model not found exception.
     *
     * @param  \Illuminate\Contracts\Queue\Job  $job
     * @param  \Throwable  $e
     * @return void
     */
    protected function handleModelNotFound(Job $job, $e)
    {
        $this->ensureUniqueJobLockIsReleasedViaContext();

        if ($job->payload()['deleteWhenMissingModels'] ?? false) {
            $this->ensureSuccessfulBatchJobIsRecordedForMissingModel($job, $job->resolveQueuedJobClass());

            return $job->delete();
        }

        return $job->fail($e);
    }

    /**
     * Ensure the lock for a unique job is released via context.
     *
     * This is required when we can't unserialize the job due to missing models.
     *
     * @return void
     */
    protected function ensureUniqueJobLockIsReleasedViaContext()
    {
        if (! $this->container->bound(ContextRepository::class) ||
            ! $this->container->bound(CacheFactory::class)) {
            return;
        }

        $context = $this->container->make(ContextRepository::class);

        [$store, $key, $owner] = [
            $context->getHidden('laravel_unique_job_cache_store'),
            $context->getHidden('laravel_unique_job_key'),
            $context->getHidden('laravel_unique_job_lock_owner'),
        ];

        if ($store && $key) {
            $cache = $this->container->make(CacheFactory::class)->store($store);

            if (is_string($owner) && $owner !== '' && $cache->getStore() instanceof LockProvider) {
                $cache->restoreLock($key, $owner)->release();
            } elseif (is_null($owner) || $owner === '') {
                $cache->lock($key)->forceRelease();
            }
        }
    }

    /**
     * Record a potentially batched job as successful when deleted because models were missing.
     *
     * @param  \Illuminate\Contracts\Queue\Job  $job
     * @param  string  $class
     * @return void
     */
    protected function ensureSuccessfulBatchJobIsRecordedForMissingModel(Job $job, string $class)
    {
        if (! isset(class_uses_recursive($class)[Batchable::class])) {
            return;
        }

        if (! $this->container->bound(BatchRepository::class)) {
            return;
        }

        $batchId = $job->payload()['data']['batchId'] ?? null;

        if ((! is_string($batchId) || $batchId === '') ||
             ! is_string($job->uuid()) || $job->uuid() === '') {
            return;
        }

        if ($batch = $this->container->make(BatchRepository::class)->find($batchId)) {
            $batch->recordSuccessfulJob($job->uuid());
        }
    }

    /**
     * Call the failed method on the job instance.
     *
     * The exception that caused the failure will be passed.
     *
     * @param  array  $data
     * @param  \Throwable|null  $e
     * @param  string  $uuid
     * @param  \Illuminate\Contracts\Queue\Job|null  $job
     * @return void
     */
    public function failed(array $data, $e, string $uuid, ?Job $job = null)
    {
        $command = $this->getCommand($data);

        if (! is_null($job)) {
            $command = $this->setJobInstanceIfNecessary($job, $command);
        }

        if (! $this->commandShouldBeUniqueUntilProcessing($command)) {
            $this->ensureUniqueJobLockIsReleased($command);
        }

        if ($command instanceof \__PHP_Incomplete_Class) {
            return;
        }

        $this->ensureFailedBatchJobIsRecorded($uuid, $command, $e);
        $this->ensureChainCatchCallbacksAreInvoked($uuid, $command, $e);

        if (method_exists($command, 'failed')) {
            $command->failed($e);
        }
    }

    /**
     * Ensure the batch is notified of the failed job.
     *
     * @param  string  $uuid
     * @param  mixed  $command
     * @param  \Throwable  $e
     * @return void
     */
    protected function ensureFailedBatchJobIsRecorded(string $uuid, $command, $e)
    {
        if (! isset(class_uses_recursive($command)[Batchable::class])) {
            return;
        }

        if ($batch = $command->batch()) {
            $batch->recordFailedJob($uuid, $e);
        }
    }

    /**
     * Ensure the chained job catch callbacks are invoked.
     *
     * @param  string  $uuid
     * @param  mixed  $command
     * @param  \Throwable  $e
     * @return void
     */
    protected function ensureChainCatchCallbacksAreInvoked(string $uuid, $command, $e)
    {
        if (method_exists($command, 'invokeChainCatchCallbacks')) {
            $command->invokeChainCatchCallbacks($e);
        }
    }

    /**
     * Get the command currently being processed.
     *
     * @return mixed
     */
    public function getRunningCommand()
    {
        return $this->runningCommand;
    }
}