fix: job concurrency (#94)

app/Jobs/Middleware/RateLimitSchedulableJobMiddleware.php

@@ -0,0 +1,33 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Jobs\Middleware;
+
+use Closure;
+use Illuminate\Support\Facades\Redis;
+
+class RateLimitSchedulableJobMiddleware
+{
+    public function __construct(
+        public string $key
+    ) {
+        //
+    }
+
+    /**
+     * Process the queued job.
+     *
+     * @param \Closure(object): void $next
+     */
+    public function handle(object $job, Closure $next): void
+    {
+        Redis::throttle('rate-limit-job:' . $this->key)
+            ->allow(1)
+            ->every(1)
+            ->then(
+                fn () => $next($job),
+                fn () => $job->release(2)
+            );
+    }
+}

app/Jobs/PersistTemporaryTableData.php

@@ -6,6 +6,7 @@
 
 use App\Contracts\ClearsCache;
 use App\Contracts\TemporaryTable;
+use App\Jobs\Middleware\RateLimitSchedulableJobMiddleware;
 use Exception;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Queue\Queueable;
@@ -69,4 +70,11 @@ public function handle(): void
             $model->clearCache($this->electionId);
         }
     }
+
+    public function middleware(): array
+    {
+        return [
+            new RateLimitSchedulableJobMiddleware('persist-temporary-table-data'),
+        ];
+    }
 }

app/Jobs/SchedulableJob.php

@@ -4,6 +4,7 @@
 
 namespace App\Jobs;
 
+use App\Jobs\Middleware\RateLimitSchedulableJobMiddleware;
 use App\Models\ScheduledJob;
 use Illuminate\Contracts\Queue\ShouldBeUnique;
 use Illuminate\Contracts\Queue\ShouldQueue;
@@ -50,6 +51,13 @@ final public function handle(): void
         $this->scheduledJob->touch('last_run_at');
     }
 
+    public function middleware(): array
+    {
+        return [
+            new RateLimitSchedulableJobMiddleware('schedulable-job'),
+        ];
+    }
+
     public function uniqueId(): string
     {
         return "scheduled-job:{$this->scheduledJob->id}";

app/Models/ScheduledJob.php

@@ -107,9 +107,6 @@ public function getPreparedSourceUrl(array $map = []): string
 
     public function fetchSource(array $map = []): Response
     {
-        // Jitter to prevent rate limiting
-        usleep(1000 * rand(50, 256));
-
         return Http::createPendingRequest()
             ->when($this->requiresAuthentication(), function (PendingRequest $request) {
                 return $request->withBasicAuth($this->source_username, $this->source_password);

config/horizon.php

@@ -205,7 +205,7 @@
         '*' => [
             'supervisor-1' => [
                 'minProcesses' => 1,
-                'maxProcesses' => 50,
+                'maxProcesses' => 20,
             ],
         ],
     ],