[11.x] Fix integrity constraint violation on failed_jobs_uuid_unique (#53230)

Author: bytestream

src/Illuminate/Queue/Console/WorkCommand.php

@@ -13,14 +13,16 @@
 use Illuminate\Queue\WorkerOptions;
 use Illuminate\Support\Carbon;
 use Illuminate\Support\InteractsWithTime;
+use SplObserver;
+use SplSubject;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Terminal;
 use Throwable;
 
 use function Termwind\terminal;
 
 #[AsCommand(name: 'queue:work')]
-class WorkCommand extends Command
+class WorkCommand extends Command implements SplObserver
 {
     use InteractsWithTime;
 
@@ -126,6 +128,27 @@ public function handle()
         );
     }
 
+    /**
+     * Receive update from subjects.
+     *
+     * @param SplSubject $subject
+     * @return void
+     */
+    public function update(SplSubject $subject): void
+    {
+        if ($subject instanceof JobProcessing) {
+            $this->writeOutput($subject->job, 'starting');
+        } elseif ($subject instanceof JobProcessed) {
+            $this->writeOutput($subject->job, 'success');
+        } elseif ($subject instanceof JobReleasedAfterException) {
+            $this->writeOutput($subject->job, 'released_after_exception');
+        } elseif ($subject instanceof JobFailed) {
+            $this->writeOutput($subject->job, 'failed', $subject->exception);
+
+            $this->logFailedJob($subject);
+        }
+    }
+
     /**
      * Run the worker instance.
      *
@@ -136,6 +159,7 @@ public function handle()
     protected function runWorker($connection, $queue)
     {
         return $this->worker
+            ->setObserver($this)
             ->setName($this->option('name'))
             ->setCache($this->cache)
             ->{$this->option('once') ? 'runNextJob' : 'daemon'}(
@@ -165,32 +189,6 @@ protected function gatherWorkerOptions()
         );
     }
 
-    /**
-     * Listen for the queue events in order to update the console output.
-     *
-     * @return void
-     */
-    protected function listenForEvents()
-    {
-        $this->laravel['events']->listen(JobProcessing::class, function ($event) {
-            $this->writeOutput($event->job, 'starting');
-        });
-
-        $this->laravel['events']->listen(JobProcessed::class, function ($event) {
-            $this->writeOutput($event->job, 'success');
-        });
-
-        $this->laravel['events']->listen(JobReleasedAfterException::class, function ($event) {
-            $this->writeOutput($event->job, 'released_after_exception');
-        });
-
-        $this->laravel['events']->listen(JobFailed::class, function ($event) {
-            $this->writeOutput($event->job, 'failed', $event->exception);
-
-            $this->logFailedJob($event);
-        });
-    }
-
     /**
      * Write the status output for the queue worker for JSON or TTY.
      *

src/Illuminate/Queue/Events/JobFailed.php

@@ -2,7 +2,7 @@
 
 namespace Illuminate\Queue\Events;
 
-class JobFailed
+class JobFailed extends QueueEvent
 {
     /**
      * The connection name.
@@ -38,5 +38,6 @@ public function __construct($connectionName, $job, $exception)
         $this->job = $job;
         $this->exception = $exception;
         $this->connectionName = $connectionName;
+        parent::__construct();
     }
 }

src/Illuminate/Queue/Events/JobProcessed.php

@@ -2,7 +2,7 @@
 
 namespace Illuminate\Queue\Events;
 
-class JobProcessed
+class JobProcessed extends QueueEvent
 {
     /**
      * The connection name.
@@ -29,5 +29,6 @@ public function __construct($connectionName, $job)
     {
         $this->job = $job;
         $this->connectionName = $connectionName;
+        parent::__construct();
     }
 }

src/Illuminate/Queue/Events/JobProcessing.php

@@ -2,7 +2,7 @@
 
 namespace Illuminate\Queue\Events;
 
-class JobProcessing
+class JobProcessing extends QueueEvent
 {
     /**
      * The connection name.
@@ -29,5 +29,6 @@ public function __construct($connectionName, $job)
     {
         $this->job = $job;
         $this->connectionName = $connectionName;
+        parent::__construct();
     }
 }

src/Illuminate/Queue/Events/JobReleasedAfterException.php

@@ -2,7 +2,7 @@
 
 namespace Illuminate\Queue\Events;
 
-class JobReleasedAfterException
+class JobReleasedAfterException extends QueueEvent
 {
     /**
      * The connection name.
@@ -29,5 +29,6 @@ public function __construct($connectionName, $job)
     {
         $this->job = $job;
         $this->connectionName = $connectionName;
+        parent::__construct();
     }
 }

src/Illuminate/Queue/Events/QueueEvent.php

@@ -0,0 +1,34 @@
+<?php
+
+namespace Illuminate\Queue\Events;
+
+use SplObjectStorage;
+use SplObserver;
+use SplSubject;
+
+abstract class QueueEvent implements SplSubject
+{
+    protected SplObjectStorage $observers;
+
+    public function __construct()
+    {
+        $this->observers = new SplObjectStorage;
+    }
+
+    public function attach(SplObserver $observer): void
+    {
+        $this->observers->attach($observer);
+    }
+
+    public function detach(SplObserver $observer): void
+    {
+        $this->observers->detach($observer);
+    }
+
+    public function notify(): void
+    {
+        foreach ($this->observers as $observer) {
+            $observer->update($this);
+        }
+    }
+}

src/Illuminate/Queue/Jobs/Job.php

@@ -9,6 +9,7 @@
 use Illuminate\Queue\ManuallyFailedException;
 use Illuminate\Queue\TimeoutExceededException;
 use Illuminate\Support\InteractsWithTime;
+use SplObserver;
 use Throwable;
 
 abstract class Job
@@ -64,6 +65,13 @@ abstract class Job
      */
     protected $queue;
 
+    /**
+     * Event observer.
+     *
+     * @var SplObserver|null
+     */
+    protected $observer;
+
     /**
      * Get the job identifier.
      *
@@ -212,9 +220,12 @@ public function fail($e = null)
 
             $this->failed($e);
         } finally {
-            $this->resolve(Dispatcher::class)->dispatch(new JobFailed(
-                $this->connectionName, $this, $e ?: new ManuallyFailedException
-            ));
+            $event = new JobFailed($this->connectionName, $this, $e ?: new ManuallyFailedException);
+            if ($this->observer) {
+                $event->attach($this->observer);
+            }
+
+            $this->resolve(Dispatcher::class)->dispatch($event);
         }
     }
 
@@ -377,4 +388,15 @@ public function getContainer()
     {
         return $this->container;
     }
+
+    /**
+     * Set the event observer.
+     *
+     * @param SplObserver $observer
+     * @return void
+     */
+    public function setObserver(SplObserver $observer)
+    {
+        $this->observer = $observer;
+    }
 }

src/Illuminate/Queue/Worker.php

@@ -18,6 +18,7 @@
 use Illuminate\Queue\Events\Looping;
 use Illuminate\Queue\Events\WorkerStopping;
 use Illuminate\Support\Carbon;
+use SplObserver;
 use Throwable;
 
 class Worker
@@ -98,6 +99,13 @@ class Worker
      */
     protected static $popCallbacks = [];
 
+    /**
+     * Attach observers to this handler to subscribe to messages.
+     *
+     * @var SplObserver|null
+     */
+    protected $observer;
+
     /**
      * Create a new queue worker.
      *
@@ -488,9 +496,12 @@ protected function handleJobException($connectionName, $job, WorkerOptions $opti
             if (! $job->isDeleted() && ! $job->isReleased() && ! $job->hasFailed()) {
                 $job->release($this->calculateBackoff($job, $options));
 
-                $this->events->dispatch(new JobReleasedAfterException(
-                    $connectionName, $job
-                ));
+                $event = new JobReleasedAfterException($connectionName, $job);
+                if ($this->observer) {
+                    $event->attach($this->observer);
+                }
+
+                $this->events->dispatch($event);
             }
         }
 
@@ -600,6 +611,10 @@ protected function markJobAsFailedIfItShouldFailOnTimeout($connectionName, $job,
      */
     protected function failJob($job, Throwable $e)
     {
+        if ($this->observer && method_exists($job, 'setObserver')) {
+            $job->setObserver($this->observer);
+        }
+
         $job->fail($e);
     }
 
@@ -656,9 +671,12 @@ protected function raiseAfterJobPopEvent($connectionName, $job)
      */
     protected function raiseBeforeJobEvent($connectionName, $job)
     {
-        $this->events->dispatch(new JobProcessing(
-            $connectionName, $job
-        ));
+        $jobProcessing = new JobProcessing($connectionName, $job);
+        if ($this->observer) {
+            $jobProcessing->attach($this->observer);
+        }
+
+        $this->events->dispatch($jobProcessing);
     }
 
     /**
@@ -670,9 +688,12 @@ protected function raiseBeforeJobEvent($connectionName, $job)
      */
     protected function raiseAfterJobEvent($connectionName, $job)
     {
-        $this->events->dispatch(new JobProcessed(
-            $connectionName, $job
-        ));
+        $event = new JobProcessed($connectionName, $job);
+        if ($this->observer) {
+            $event->attach($this->observer);
+        }
+
+        $this->events->dispatch($event);
     }
 
     /**
@@ -880,4 +901,17 @@ public function setManager(QueueManager $manager)
     {
         $this->manager = $manager;
     }
+
+    /**
+     * Attach an observer.
+     *
+     * @param SplObserver $observer
+     * @return $this
+     */
+    public function setObserver(SplObserver $observer)
+    {
+        $this->observer = $observer;
+
+        return $this;
+    }
 }