wip

Author: Marick van Tuil

src/Worker.php

@@ -4,8 +4,11 @@
 
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
+use Illuminate\Contracts\Debug\ExceptionHandler;
+use Illuminate\Queue\Events\JobTimedOut;
 use Illuminate\Queue\Worker as LaravelWorker;
 use Illuminate\Queue\WorkerOptions;
+use Symfony\Component\ErrorHandler\Error\FatalError;
 
 /**
  * Custom worker class to handle specific requirements for Google Cloud Tasks.
@@ -14,52 +17,57 @@
  * integrate with Google Cloud Tasks, particularly focusing on job timeout
  * handling and graceful shutdowns to avoid interrupting the HTTP lifecycle.
  *
- * Firstly, the 'supportsAsyncSignals', 'listenForSignals', and 'registerTimeoutHandler' methods
- * are protected and called within the queue while(true) loop. We want (and need!) to have that
- * too in order to support job timeouts. So, to make it work, we create a public method that
- * can call the private signal methods.
- *
- * Secondly, we need to override the 'kill' method because it tends to kill the server process (artisan serve, octane),
- * as well as abort the HTTP request from Cloud Tasks. This is not the desired behavior.
- * Instead, it should just fire the WorkerStopped event and return a normal status code.
+ * Firstly, normally job timeouts are handled using the pcntl extension. Since we
+ * are running in an HTTP environment, we can't use those functions. An alternative
+ * method is using set_time_limit and when PHP throws the fatal 'Maximum execution time exceeded' error,
+ * we will handle the job error like how Laravel would if the pcntl alarm had gone off.
  */
 class Worker extends LaravelWorker
 {
     public function process($connectionName, $job, WorkerOptions $options): void
     {
-        if ($this->supportsAsyncSignals()) {
-            $this->listenForSignals();
+        assert($job instanceof CloudTasksJob);
 
-            pcntl_signal(SIGSEGV, fn () => $this->shouldQuit = true);
+        set_time_limit(max($this->timeoutForJob($job, $options), 0));
 
-            $this->registerTimeoutHandler($job, $options);
-        }
+        app(ExceptionHandler::class)->reportable(
+            fn (FatalError $error) => $this->onFatalError($error, $job, $options)
+        );
 
         parent::process($connectionName, $job, $options);
+    }
+
+    private function onFatalError(FatalError $error, CloudTasksJob $job, WorkerOptions $options): bool
+    {
+        if (fnmatch('Maximum execution time * exceeded', $error->getMessage())) {
+            $this->onJobTimedOut($job, $options);
 
-        if ($this->supportsAsyncSignals()) {
-            $this->resetTimeoutHandler();
+            return false;
         }
+
+        return true;
     }
 
-    public function kill($status = 0, $options = null): void
+    private function onJobTimedOut(CloudTasksJob $job, WorkerOptions $options): void
     {
-        if ($this->supportsAsyncSignals()) {
-            $this->resetTimeoutHandler();
-        }
+        $this->markJobAsFailedIfWillExceedMaxAttempts(
+            $job->getConnectionName(), $job, (int) $options->maxTries, $e = $this->timeoutExceededException($job)
+        );
 
-        parent::stop($status, $options);
+        $this->markJobAsFailedIfWillExceedMaxExceptions(
+            $job->getConnectionName(), $job, $e
+        );
 
-        // When running tests, we cannot run exit because it will kill the PHPunit process.
-        // So, to still test that the application has exited, we will simply rely on the
-        // WorkerStopped event that is fired when the worker is stopped.
-        if (! app()->runningUnitTests()) {
-            if (extension_loaded('posix') && extension_loaded('pcntl')) {
-                posix_kill(getmypid(), SIGKILL);
-            }
+        $this->markJobAsFailedIfItShouldFailOnTimeout(
+            $job->getConnectionName(), $job, $e
+        );
 
-            exit($status);
-        }
+        $this->events->dispatch(new JobTimedOut(
+            $job->getConnectionName(), $job
+        ));
 
+        if (! $job->isDeleted() && ! $job->isReleased() && ! $job->hasFailed()) {
+            $job->release($this->calculateBackoff($job, $options));
+        }
     }
 }