Feature/failed jobs table

CHANGELOG.md

@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## 2.1.0-beta1 - 2021-03-28
+
+**Added**
+
+- Handling of failed jobs
+
 ## 2.0.1 - 2020-12-06
 
 **Fixed**

phpunit.xml

@@ -23,6 +23,7 @@
         <env name="MAIL_DRIVER" value="log"/>
         <env name="QUEUE_DRIVER" value="cloudtasks"/>
         <env name="GOOGLE_APPLICATION_CREDENTIALS" value="./tests/Support/gcloud-key-valid.json" />
+        <env name="DB_CONNECTION" value="sqlite" />
     </php>
 
     <filter>

src/CloudTasksJob.php

@@ -10,11 +10,11 @@ class CloudTasksJob extends LaravelJob implements JobContract
 {
     private $job;
     private $attempts;
+    private $maxTries;
 
-    public function __construct($job, $attempts)
+    public function __construct($job)
     {
         $this->job = $job;
-        $this->attempts = $attempts;
         $this->container = Container::getInstance();
     }
 
@@ -32,4 +32,28 @@ public function attempts()
     {
         return $this->attempts;
     }
+
+    public function setAttempts($attempts)
+    {
+        $this->attempts = $attempts;
+    }
+
+    public function setMaxTries($maxTries)
+    {
+        if ((int) $maxTries === -1) {
+            $maxTries = null;
+        }
+
+        $this->maxTries = $maxTries;
+    }
+
+    public function maxTries()
+    {
+        return $this->maxTries;
+    }
+
+    public function setQueue($queue)
+    {
+        $this->queue = $queue;
+    }
 }

src/TaskHandler.php

@@ -4,22 +4,19 @@
 
 use Google\Cloud\Tasks\V2\CloudTasksClient;
 use Illuminate\Http\Request;
+use Illuminate\Queue\Events\JobFailed;
 use Illuminate\Queue\Worker;
 use Illuminate\Queue\WorkerOptions;
-use Firebase\JWT\JWT;
 
 class TaskHandler
 {
     private $request;
-    private $guzzle;
-    private $jwt;
     private $publicKey;
 
-    public function __construct(CloudTasksClient $client, Request $request, JWT $jwt, OpenIdVerificator $publicKey)
+    public function __construct(CloudTasksClient $client, Request $request, OpenIdVerificator $publicKey)
     {
         $this->client = $client;
         $this->request = $request;
-        $this->jwt = $jwt;
         $this->publicKey = $publicKey;
     }
 
@@ -33,6 +30,8 @@ public function handle($task = null)
 
         $task = $task ?: $this->captureTask();
 
+        $this->listenForEvents();
+
         $this->handleTask($task);
     }
 
@@ -81,7 +80,11 @@ private function captureTask()
     {
         $input = file_get_contents('php://input');
 
-        if ($input === false) {
+        if (!$input) {
+            $input = request('input') ?: false;
+        }
+
+        if (!$input) {
             throw new CloudTasksException('Could not read incoming task');
         }
 
@@ -94,19 +97,47 @@ private function captureTask()
         return $task;
     }
 
+    private function listenForEvents()
+    {
+        app('events')->listen(JobFailed::class, function ($event) {
+            app('queue.failer')->log(
+                'cloudtasks', $event->job->getQueue(),
+                $event->job->getRawBody(), $event->exception
+            );
+        });
+    }
+
     /**
      * @param $task
      * @throws CloudTasksException
      */
     private function handleTask($task)
     {
-        $job = new CloudTasksJob($task, request()->header('X-CloudTasks-TaskRetryCount'));
+        $job = new CloudTasksJob($task);
+
+        $job->setAttempts(request()->header('X-CloudTasks-TaskRetryCount') + 1);
+        $job->setQueue(request()->header('X-Cloudtasks-Queuename'));
+        $job->setMaxTries($this->getQueueMaxTries($job));
 
         $worker = $this->getQueueWorker();
 
         $worker->process('cloudtasks', $job, new WorkerOptions());
     }
 
+    private function getQueueMaxTries(CloudTasksJob $job)
+    {
+        $queueName = $this->client->queueName(
+            Config::project(),
+            Config::location(),
+            $job->getQueue()
+        );
+
+        return $this->client
+            ->getQueue($queueName)
+            ->getRetryConfig()
+            ->getMaxAttempts();
+    }
+
     /**
      * @return Worker
      */

tests/GooglePublicKeyTest.php

@@ -8,7 +8,6 @@
 use Illuminate\Cache\Events\CacheHit;
 use Illuminate\Cache\Events\CacheMissed;
 use Illuminate\Cache\Events\KeyWritten;
-use Illuminate\Support\Facades\Cache;
 use Illuminate\Support\Facades\Event;
 use Mockery;
 use phpseclib\Crypt\RSA;

tests/Support/FailingJob.php

@@ -0,0 +1,34 @@
+<?php
+
+namespace Tests\Support;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Foundation\Bus\Dispatchable;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Queue\SerializesModels;
+
+class FailingJob implements ShouldQueue
+{
+    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
+
+    /**
+     * Create a new job instance.
+     *
+     * @return void
+     */
+    public function __construct()
+    {
+        //
+    }
+
+    /**
+     * Execute the job.
+     *
+     * @return void
+     */
+    public function handle()
+    {
+        throw new \Error('simulating a failing job');
+    }
+}

tests/Support/failing-job-payload.json

@@ -0,0 +1 @@
+{"uuid":"f4e1ea03-3ab9-45f8-a4a1-50218169472e","displayName":"Tests\\Support\\FailingJob","job":"Illuminate\\Queue\\CallQueuedHandler@call","maxTries":null,"maxExceptions":null,"delay":null,"timeout":null,"timeoutAt":null,"data":{"commandName":"Tests\\Support\\FailingJob","command":"O:24:\"Tests\\Support\\FailingJob\":8:{s:3:\"job\";N;s:10:\"connection\";N;s:5:\"queue\";N;s:15:\"chainConnection\";N;s:10:\"chainQueue\";N;s:5:\"delay\";N;s:10:\"middleware\";a:0:{}s:7:\"chained\";a:0:{}}"}}

tests/TaskHandlerTest.php

@@ -7,6 +7,7 @@
 use Google\Cloud\Tasks\V2\CloudTasksClient;
 use Illuminate\Cache\Events\CacheHit;
 use Illuminate\Cache\Events\KeyWritten;
+use Illuminate\Support\Facades\DB;
 use Illuminate\Support\Facades\Event;
 use Illuminate\Support\Facades\Mail;
 use Mockery;
@@ -45,10 +46,23 @@ protected function setUp(): void
         $googlePublicKey->shouldReceive('getPublicKey')->andReturnNull();
         $googlePublicKey->shouldReceive('getKidFromOpenIdToken')->andReturnNull();
 
+        $cloudTasksClient = Mockery::mock(new CloudTasksClient());
+
+        // Ensure we don't fetch the Queue name and attempts each test...
+        $cloudTasksClient->shouldReceive('queueName')->andReturn('my-queue');
+        $cloudTasksClient->shouldReceive('getQueue')->andReturn(new class {
+            public function getRetryConfig() {
+                return new class {
+                    public function getMaxAttempts() {
+                        return 3;
+                    }
+                };
+            }
+        });
+
         $this->handler = new TaskHandler(
-            new CloudTasksClient(),
+            $cloudTasksClient,
             request(),
-            $this->jwt,
             $googlePublicKey
         );
 
@@ -66,21 +80,6 @@ public function it_needs_an_authorization_header()
         $this->handler->handle();
     }
 
-    /** @test */
-    public function the_authorization_header_must_contain_a_valid_gcloud_token()
-    {
-        request()->headers->add([
-            'Authorization' => 'Bearer 123',
-        ]);
-
-        $this->expectException(CloudTasksException::class);
-        $this->expectExceptionMessage('Could not decode incoming task');
-
-        $this->handler->handle();
-
-        // @todo - test with a valid token, not sure how to do that right now
-    }
-
     /** @test */
     public function it_will_validate_the_token_iss()
     {
@@ -144,8 +143,46 @@ public function it_runs_the_incoming_job()
         Mail::assertSent(TestMailable::class);
     }
 
+    /** @test */
+    public function after_max_attempts_it_will_log_to_failed_table()
+    {
+        $this->request->headers->add(['X-Cloudtasks-Queuename' => 'my-queue']);
+
+        $this->request->headers->add(['X-CloudTasks-TaskRetryCount' => 1]);
+        try {
+            $this->handler->handle($this->failingJob());
+        } catch (\Throwable $e) {
+            //
+        }
+
+        $this->assertCount(0, DB::table('failed_jobs')->get());
+
+        $this->request->headers->add(['X-CloudTasks-TaskRetryCount' => 2]);
+        try {
+            $this->handler->handle($this->failingJob());
+        } catch (\Throwable $e) {
+            //
+        }
+
+        $this->assertDatabaseHas('failed_jobs', [
+            'connection' => 'cloudtasks',
+            'queue' => 'my-queue',
+            'payload' => rtrim($this->failingJobPayload()),
+        ]);
+    }
+
     private function simpleJob()
     {
         return json_decode(file_get_contents(__DIR__ . '/Support/test-job-payload.json'), true);
     }
+
+    private function failingJobPayload()
+    {
+        return file_get_contents(__DIR__ . '/Support/failing-job-payload.json');
+    }
+
+    private function failingJob()
+    {
+        return json_decode($this->failingJobPayload(), true);
+    }
 }

tests/TestCase.php

@@ -2,10 +2,33 @@
 
 namespace Tests;
 
-use Illuminate\Support\Facades\Artisan;
-
 class TestCase extends \Orchestra\Testbench\TestCase
 {
+    public static $migrated = false;
+
+    protected function setUp(): void
+    {
+        parent::setUp();
+
+        // There is probably a more sane way to do this
+        if (!static::$migrated) {
+            if (file_exists(database_path('database.sqlite'))) {
+                unlink(database_path('database.sqlite'));
+            }
+
+            touch(database_path('database.sqlite'));
+
+            foreach(glob(database_path('migrations/*.php')) as $file) {
+                unlink($file);
+            }
+
+            $this->artisan('queue:failed-table');
+            $this->artisan('migrate');
+
+            static::$migrated = true;
+        }
+    }
+
     /**
      * Get package providers.  At a minimum this is the package being tested, but also
      * would include packages upon which our package depends, e.g. Cartalyst/Sentry