Merge branch '4.x-dev' into 4.x-dev

Author: aaajeetee

src/CloudTasksQueue.php

@@ -24,7 +24,7 @@
 
 class CloudTasksQueue extends LaravelQueue implements QueueContract
 {
-    private Closure | array $headers = [];
+    private Closure|array $headers = [];
     private static ?Closure $handlerUrlCallback = null;
 
     public function __construct(public array $config, public CloudTasksClient $client, public $dispatchAfterCommit = false)
@@ -160,7 +160,7 @@ private function taskName(string $queueName, array $payload): string
             $this->config['project'],
             $this->config['location'],
             $queueName,
-            $displayName.'-'. bin2hex(random_bytes(8)),
+            $displayName.'-'.bin2hex(random_bytes(8)),
         );
     }
 
@@ -180,8 +180,7 @@ private function enrichPayloadWithInternalData(
         string $queueName,
         string $taskName,
         string $connectionName,
-    ): array
-    {
+    ): array {
         $payload['internal'] = [
             'attempts' => $payload['internal']['attempts'] ?? 0,
             'queue' => $queueName,
@@ -197,7 +196,7 @@ public function addPayloadToTask(array $payload, Task $task, mixed $job): Task
     {
         $headers = value($this->headers, $payload) ?: [];
 
-        if (!empty($this->config['app_engine'])) {
+        if (! empty($this->config['app_engine'])) {
             $path = \Safe\parse_url(route('cloud-tasks.handle-task'), PHP_URL_PATH);
 
             $appEngineRequest = new AppEngineHttpRequest();
@@ -206,7 +205,7 @@ public function addPayloadToTask(array $payload, Task $task, mixed $job): Task
             $appEngineRequest->setBody(json_encode($payload));
             $appEngineRequest->setHeaders($headers);
 
-            if (!empty($service = $this->config['app_engine_service'])) {
+            if (! empty($service = $this->config['app_engine_service'])) {
                 $routing = new AppEngineRouting();
                 $routing->setService($service);
                 $appEngineRequest->setAppEngineRouting($routing);
@@ -264,14 +263,14 @@ public function getHandler(mixed $job): string
 
         $handler = rtrim($this->config['handler'], '/');
 
-        if (str_ends_with($handler, '/'. config('cloud-tasks.uri'))) {
+        if (str_ends_with($handler, '/'.config('cloud-tasks.uri'))) {
             return $handler;
         }
 
-        return $handler.'/'. config('cloud-tasks.uri');
+        return $handler.'/'.config('cloud-tasks.uri');
     }
 
-    public function setTaskHeaders(Closure | array $headers): void
+    public function setTaskHeaders(Closure|array $headers): void
     {
         $this->headers = $headers;
     }

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/IncomingTask.php

@@ -5,6 +5,7 @@
 namespace Stackkit\LaravelGoogleCloudTasksQueue;
 
 use Safe\Exceptions\JsonException;
+
 use function Safe\json_decode;
 
 class IncomingTask

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/QueueTest.php

@@ -16,7 +16,6 @@
 use Illuminate\Support\Facades\Queue;
 use PHPUnit\Framework\Attributes\Test;
 use Stackkit\LaravelGoogleCloudTasksQueue\CloudTasksApi;
-use Stackkit\LaravelGoogleCloudTasksQueue\CloudTasksQueue;
 use Stackkit\LaravelGoogleCloudTasksQueue\Events\JobReleased;
 use Tests\Support\FailingJob;
 use Tests\Support\FailingJobWithExponentialBackoff;

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
 {
@@ -37,7 +38,7 @@ public function it_can_run_a_task()
         $this->dispatch(new SimpleJob())->runWithoutExceptionHandler();
 
         // Assert
-        Event::assertDispatched(fn(JobOutput $event) => $event->output === 'SimpleJob:success');
+        Event::assertDispatched(fn (JobOutput $event) => $event->output === 'SimpleJob:success');
     }
 
     #[Test]
@@ -54,7 +55,7 @@ public function it_can_run_a_task_using_the_task_connection()
         $this->dispatch($job)->runWithoutExceptionHandler();
 
         // Assert
-        Event::assertDispatched(fn(JobOutput $event) => $event->output === 'SimpleJob:success');
+        Event::assertDispatched(fn (JobOutput $event) => $event->output === 'SimpleJob:success');
     }
 
     #[Test]
@@ -185,7 +186,7 @@ public function it_can_handle_encrypted_jobs()
             decrypt($job->payloadAsArray('data.command')),
         );
 
-        Event::assertDispatched(fn(JobOutput $event) => $event->output === 'EncryptedJob:success');
+        Event::assertDispatched(fn (JobOutput $event) => $event->output === 'EncryptedJob:success');
     }
 
     #[Test]
@@ -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);
+    }
 }