Prevent multithread jobs

emodyz · Aug 13, 2022 · 181b778 · 181b778
1 parent 0986274
commit 181b778
Show file tree

Hide file tree

Showing 4 changed files with 49 additions and 56 deletions.
diff --git a/app/Http/Controllers/ModPackController.php b/app/Http/Controllers/ModPackController.php
@@ -125,7 +125,7 @@ public function edit(Request $request, Modpack $modpack): Response|ModPackResour
           ->get()
           ->map(fn(Server $server) => $server->only(['id', 'name', 'logo_url', 'game']));
 
-        $modpack->servers;
+        $modpack->load('servers');
 
         return inertia('ModPacks/Edit', compact('servers', 'modpack'));
     }
@@ -209,6 +209,11 @@ public function cancelUpdate(Modpack $modpack): Response
         }
 
         $modpack->batch->cancel();
+
+        $modpack->update([
+            'job_batch_id' => null,
+        ]);
+
         return response()->noContent();
     }
 }
diff --git a/app/Jobs/ProcessModPackFile.php b/app/Jobs/ProcessModPackFile.php
@@ -9,8 +9,8 @@
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
-use Illuminate\Queue\Middleware\WithoutOverlapping;
 use Illuminate\Queue\SerializesModels;
+use Illuminate\Support\Facades\Redis;
 use Illuminate\Support\Facades\Storage;
 use Illuminate\Support\Str;
 use Throwable;
@@ -28,24 +28,15 @@ class ProcessModPackFile implements ShouldQueue
     /**
      * Create a new job instance.
      *
-     * @param Modpack $modpack
-     * @param string $filePath
+     * @param  Modpack  $modpack
+     * @param  string  $filePath
      */
     public function __construct(Modpack $modpack, string $filePath)
     {
         $this->modpack = $modpack;
         $this->filePath = $filePath;
     }
 
-    /**
-     * Get the middleware the job should pass through.
-     *
-     * @return array
-     */
-    public function middleware()
-    {
-        return [new WithoutOverlapping($this->modpack->id)];
-    }
 
     /**
      * Execute the job.
@@ -55,36 +46,39 @@ public function middleware()
      */
     public function handle()
     {
-        if ($this->batch()->cancelled()) {
+        if ($this->batch()->cancelled() || $this->batch()->hasFailures()) {
+            info('Batch canceled or has failures, file skipped '. $this->filePath);
             return;
         }
 
+        info('Processing file '. $this->filePath);
+
         $disk = $this->modpack->disk;
 
         $filePath = Storage::disk($disk)->path($this->filePath);
         $fileSize = Storage::disk($disk)->size($this->filePath);
         $fileUrl = Storage::disk($disk)->url($this->filePath);
         $fileHash = hash_file('sha256', $filePath);
         $fileName = basename($filePath);
-        $filePath = (string)Str::of($this->filePath)->replaceFirst(
+        $filePath = (string) Str::of($this->filePath)->replaceFirst(
             $this->modpack->path,
             $this->modpack->name
         );
         $filePathPrevented = Str::of($filePath)
             ->replace('.', '-');
 
-        $this->modpack->forceFill([
-            "manifest_info->size" => $this->modpack->manifest_info['size'] + $fileSize,
-            "manifest_info->files" => $this->modpack->manifest_info['files'] + 1,
-            "manifest->{$filePathPrevented}" => [
-                'url' => $fileUrl,
-                'size' => $fileSize,
-                'name' => $fileName,
-                'path' => $filePath,
-                'sha256' => $fileHash
-            ]
-        ])->saveOrFail();
+        Redis::hIncrBy("modpackManifestInfoUpdate:{$this->modpack->id}", 'size', $fileSize);
+        Redis::hIncrBy("modpackManifestInfoUpdate:{$this->modpack->id}", 'files', 1);
+
+        Redis::hSet("modpackManifestUpdate:{$this->modpack->id}", $filePathPrevented, json_encode([
+            'url' => $fileUrl,
+            'size' => $fileSize,
+            'name' => $fileName,
+            'path' => $filePath,
+            'sha256' => $fileHash
+        ]));
 
         ModPackProcessProgress::broadcast($this->modpack, $this->batch()->progress());
     }
 }
+
diff --git a/app/Listeners/Modpack/StartModPackUpdate.php b/app/Listeners/Modpack/StartModPackUpdate.php
@@ -8,12 +8,16 @@
 use App\Events\ModPack\ModPackProcessStarted;
 use App\Events\ModPack\ModPackUpdateRequested;
 use App\Jobs\ProcessModPackFile;
+use Exception;
 use Illuminate\Bus\Batch;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Support\Facades\Bus;
+use Illuminate\Support\Facades\Log;
+use Illuminate\Support\Facades\Redis;
 use Illuminate\Support\Facades\Storage;
 use Throwable;
+use function React\Promise\map;
 
 class StartModPackUpdate implements ShouldQueue
 {
@@ -29,37 +33,41 @@ class StartModPackUpdate implements ShouldQueue
     public function handle(ModPackUpdateRequested $event)
     {
         $modpack = $event->modPack;
-        $savedManifest = $modpack->manifest;
-        $savedManifestInfo = $modpack->manifest_info;
-        $modpack->cleanManifest();
 
         $files = collect(Storage::disk($modpack->disk)->files($modpack->path, true));
         if ($files->isEmpty()) {
             ModPackProcessDone::broadcast($modpack);
             return true;
         }
 
+        Redis::del("modpackManifestUpdate:$modpack->id");
+        Redis::del("modpackManifestInfoUpdate:$modpack->id");
+
         $jobs = $files->map(fn($file) => new ProcessModPackFile($modpack, $file));
         $batch = Bus::batch($jobs->toArray())
-            ->then(function (Batch $batch) use ($modpack, $savedManifest, $savedManifestInfo) {
+            ->then(function (Batch $batch) use ($modpack) {
                 if ($batch->canceled()) {
+                    ModPackProcessCanceled::broadcast($modpack);
+                    return;
+                }
+                $manifest = Redis::hGetAll("modpackManifestUpdate:$modpack->id");
+                $manifestInfo = Redis::hGetAll("modpackManifestInfoUpdate:$modpack->id");
+
+                if (!empty($manifest) && !empty($manifestInfo)) {
                     $modpack->update([
-                        'manifest' => $savedManifest,
-                        'manifest_info' => $savedManifestInfo
+                        'manifest' => collect($manifest)->map(fn ($value) => json_decode($value)),
+                        'manifest_info' => $manifestInfo,
+                        'manifest_last_update' => now()->toDateTimeString()
                     ]);
-                    ModPackProcessCanceled::broadcast($modpack);
+                    ModPackProcessDone::broadcast($modpack);
+                    Log::info('Job done');
                     return;
                 }
-                $modpack->update([
-                    'manifest_last_update' => now()->toDateTimeString()
-                ]);
-                ModPackProcessDone::broadcast($modpack);
-            })->catch(function () use ($modpack, $savedManifest, $savedManifestInfo) {
-                $modpack->update([
-                    'manifest' => $savedManifest,
-                    'manifest_info' => $savedManifestInfo
-                ]);
+
+                throw new Exception('Job not finalized correctly...');
+            })->catch(function () use ($modpack) {
                 ModPackProcessFailed::broadcast($modpack);
+                Log::alert('Job failed');
             })->finally(function () use ($modpack) {
                 $modpack->update([
                     'job_batch_id' => null,

diff --git a/app/Models/Modpack.php b/app/Models/Modpack.php
@@ -89,18 +89,4 @@ public function getBatchAttribute(): ?Batch
         }
         return Bus::findBatch($this->job_batch_id);
     }
-
-    /**
-     * void
-     */
-    public function cleanManifest()
-    {
-        $this->update([
-            'manifest' => [],
-            'manifest_info' => [
-                "size" => 0,
-                "files" => 0
-            ]
-        ]);
-    }
 }