Update NewStatusPipeline, improve fallback

app/Jobs/StatusPipeline/NewStatusPipeline.php

@@ -4,14 +4,12 @@ namespace App\Jobs\StatusPipeline;
 
 use App\Media;
 use App\Status;
-use Cache;
 use Illuminate\Bus\Queueable;
 use Illuminate\Contracts\Queue\ShouldQueue;
 use Illuminate\Foundation\Bus\Dispatchable;
 use Illuminate\Queue\InteractsWithQueue;
 use Illuminate\Queue\SerializesModels;
 use Illuminate\Support\Facades\Log;
-use Illuminate\Support\Facades\Redis;
 
 class NewStatusPipeline implements ShouldQueue
 {
@@ -64,17 +62,10 @@ class NewStatusPipeline implements ShouldQueue
      */
     public function handle()
     {
-        // Check if status still exists
-        if (!Status::where('id', $this->status->id)->exists()) {
-            if(config('federation.activitypub.delivery.logger.enabled')) {
-                Log::info('Status ' . $this->status->id . ' was deleted before federation');
-            }
-            return;
-        }
-
         // Skip media check if cloud storage isn't enabled or fast processing is on
-        if (!config_cache('pixelfed.cloud_storage') || config('pixelfed.media_fast_process')) {
+        if (! config_cache('pixelfed.cloud_storage') || config('pixelfed.media_fast_process')) {
             $this->dispatchFederation();
+
             return;
         }
 
@@ -91,16 +82,18 @@ class NewStatusPipeline implements ShouldQueue
                 ->first();
 
             // If media has been processing for more than 10 minutes, proceed anyway
-            if ($oldestProcessingMedia && now()->diffInMinutes($oldestProcessingMedia->created_at) > 10) {
-                if(config('federation.activitypub.delivery.logger.enabled')) {
-                    Log::warning('Media processing timeout for status ' . $this->status->id . '. Proceeding with federation.');
+            if ($oldestProcessingMedia && $oldestProcessingMedia->replicated_at && $oldestProcessingMedia->replicated_at->diffInMinutes(now()) > 10) {
+                if (config('federation.activitypub.delivery.logger.enabled')) {
+                    Log::warning('Media processing timeout for status '.$this->status->id.'. Proceeding with federation.');
                 }
                 $this->dispatchFederation();
+
                 return;
             }
 
             // Release job back to queue with delay of 30 seconds
             $this->release(30);
+
             return;
         }
 
@@ -118,8 +111,8 @@ class NewStatusPipeline implements ShouldQueue
         try {
             StatusEntityLexer::dispatch($this->status);
         } catch (\Exception $e) {
-            if(config('federation.activitypub.delivery.logger.enabled')) {
-                Log::error('Federation dispatch failed for status ' . $this->status->id . ': ' . $e->getMessage());
+            if (config('federation.activitypub.delivery.logger.enabled')) {
+                Log::error('Federation dispatch failed for status '.$this->status->id.': '.$e->getMessage());
             }
             throw $e;
         }
@@ -128,15 +121,14 @@ class NewStatusPipeline implements ShouldQueue
     /**
      * Handle a job failure.
      *
-     * @param  \Throwable  $exception
      * @return void
      */
     public function failed(\Throwable $exception)
     {
-        if(config('federation.activitypub.delivery.logger.enabled')) {
-            Log::error('NewStatusPipeline failed for status ' . $this->status->id, [
+        if (config('federation.activitypub.delivery.logger.enabled')) {
+            Log::error('NewStatusPipeline failed for status '.$this->status->id, [
                 'exception' => $exception->getMessage(),
-                'trace' => $exception->getTraceAsString()
+                'trace' => $exception->getTraceAsString(),
             ]);
         }
     }

app/Media.php

@@ -22,7 +22,8 @@ class Media extends Model
     protected $casts = [
         'srcset' => 'array',
         'deleted_at' => 'datetime',
-        'skip_optimize' => 'boolean'
+        'skip_optimize' => 'boolean',
+        'replicated_at' => 'datetime',
     ];
 
     public function status()