Add support for job timeouts

Author: marickvantuil

src/CloudTasksServiceProvider.php

@@ -5,6 +5,8 @@
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
 use Google\Cloud\Tasks\V2\Client\CloudTasksClient;
+use Illuminate\Contracts\Debug\ExceptionHandler;
+use Illuminate\Foundation\Application;
 use Illuminate\Queue\Events\JobExceptionOccurred;
 use Illuminate\Queue\Events\JobFailed;
 use Illuminate\Queue\Events\JobProcessed;
@@ -29,6 +31,15 @@ private function registerClient(): void
             return new CloudTasksClient();
         });
 
+        $this->app->singleton('cloud-tasks.worker', function (Application $app) {
+            return new Worker(
+                $app['queue'],
+                $app['events'],
+                $app[ExceptionHandler::class],
+                fn() => $app->isDownForMaintenance(),
+            );
+        });
+
         $this->app->bind('cloud-tasks-api', CloudTasksApiConcrete::class);
     }
 

src/TaskHandler.php

@@ -6,7 +6,6 @@
 
 use Google\Cloud\Tasks\V2\Client\CloudTasksClient;
 use Illuminate\Container\Container;
-use Illuminate\Queue\Worker;
 use Illuminate\Queue\WorkerOptions;
 
 class TaskHandler
@@ -51,7 +50,7 @@ private function run(IncomingTask $task): void
 
         $job->setAttempts($job->attempts() + 1);
 
-        tap(app('queue.worker'), fn (Worker $worker) => $worker->process(
+        tap(app('cloud-tasks.worker'), fn (Worker $worker) => $worker->process(
             connectionName: $job->getConnectionName(),
             job: $job,
             options: $this->getWorkerOptions()

src/Worker.php

@@ -0,0 +1,52 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Stackkit\LaravelGoogleCloudTasksQueue;
+
+use Illuminate\Queue\Worker as LaravelWorker;
+use Illuminate\Queue\WorkerOptions;
+
+/**
+ * Custom worker class to handle specific requirements for Google Cloud Tasks.
+ *
+ * This class modifies the behavior of the Laravel queue worker to better
+ * 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.
+ */
+class Worker extends LaravelWorker
+{
+    public function process($connectionName, $job, WorkerOptions $options)
+    {
+        if ($this->supportsAsyncSignals()) {
+            $this->listenForSignals();
+
+            $this->registerTimeoutHandler($job, $options);
+        }
+
+        return parent::process($connectionName, $job, $options);
+    }
+
+    public function kill($status = 0, $options = null): void
+    {
+        parent::stop($status, $options);
+
+        // 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()) {
+            return;
+        }
+
+        exit($status);
+    }
+}

tests/Support/SimpleJobWithTimeout.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Support;
+
+use Illuminate\Queue\Events\WorkerStopping;
+use Illuminate\Support\Facades\Event;
+
+class SimpleJobWithTimeout extends SimpleJob
+{
+    public $timeout = 3;
+
+    public function handle()
+    {
+        Event::listen(WorkerStopping::class, function () {
+            event(new JobOutput('SimpleJobWithTimeout:worker-stopping'));
+        });
+
+        event(new JobOutput('SimpleJobWithTimeout:1'));
+        sleep(1);
+        event(new JobOutput('SimpleJobWithTimeout:2'));
+        sleep(1);
+        event(new JobOutput('SimpleJobWithTimeout:3'));
+        sleep(1);
+        event(new JobOutput('SimpleJobWithTimeout:4'));
+        sleep(1);
+        event(new JobOutput('SimpleJobWithTimeout:5'));
+    }
+}

tests/TaskHandlerTest.php

@@ -17,6 +17,7 @@
 use Tests\Support\FailingJobWithUnlimitedTries;
 use Tests\Support\JobOutput;
 use Tests\Support\SimpleJob;
+use Tests\Support\SimpleJobWithTimeout;
 
 class TaskHandlerTest extends TestCase
 {
@@ -248,4 +249,25 @@ public function retried_jobs_get_a_new_name()
         $this->assertCount(2, $this->createdTasks);
         $this->assertNotEquals($this->createdTasks[0]->getName(), $this->createdTasks[1]->getName());
     }
+
+    #[Test]
+    public function test_job_timeout()
+    {
+        // Arrange
+        Event::fake(JobOutput::class);
+
+        // Act
+        $this->dispatch(new SimpleJobWithTimeout())->run();
+
+        // Assert
+        $events = Event::dispatched(JobOutput::class)->map(fn($event) => $event[0]->output)->toArray();
+        $this->assertEquals([
+            'SimpleJobWithTimeout:1',
+            'SimpleJobWithTimeout:2',
+            'SimpleJobWithTimeout:3',
+            'SimpleJobWithTimeout:worker-stopping',
+            'SimpleJobWithTimeout:4',
+            'SimpleJobWithTimeout:5',
+        ], $events);
+    }
 }