Merge pull request #5726 from pixelfed/staging

Staging

app/Jobs/StatusPipeline/NewStatusPipeline.php

@@ -2,21 +2,21 @@
 
 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\Redis;
+use Illuminate\Support\Facades\Log;
 
 class NewStatusPipeline implements ShouldQueue
 {
     use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
 
     protected $status;
-    
+
     /**
      * Delete the job if its models no longer exist.
      *
@@ -24,9 +24,27 @@ class NewStatusPipeline implements ShouldQueue
      */
     public $deleteWhenMissingModels = true;
 
-    public $timeout = 5;
-    public $tries = 1;
-    
+    /**
+     * Increased timeout to handle cloud storage operations
+     *
+     * @var int
+     */
+    public $timeout = 30;
+
+    /**
+     * Number of times to attempt the job
+     *
+     * @var int
+     */
+    public $tries = 3;
+
+    /**
+     * Backoff periods between retries (in seconds)
+     *
+     * @var array
+     */
+    public $backoff = [30, 60, 120];
+
     /**
      * Create a new job instance.
      *
@@ -44,6 +62,74 @@ class NewStatusPipeline implements ShouldQueue
      */
     public function handle()
     {
-        StatusEntityLexer::dispatch($this->status);
+        // 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')) {
+            $this->dispatchFederation();
+
+            return;
+        }
+
+        // Check for media still processing
+        $stillProcessing = Media::whereStatusId($this->status->id)
+            ->whereNull('cdn_url')
+            ->exists();
+
+        if ($stillProcessing) {
+            // Get the oldest processing media item
+            $oldestProcessingMedia = Media::whereStatusId($this->status->id)
+                ->whereNull('cdn_url')
+                ->oldest()
+                ->first();
+
+            // If media has been processing for more than 10 minutes, proceed anyway
+            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;
+        }
+
+        // All media processed, proceed with federation
+        $this->dispatchFederation();
+    }
+
+    /**
+     * Dispatch the federation job
+     *
+     * @return void
+     */
+    protected function dispatchFederation()
+    {
+        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());
+            }
+            throw $e;
+        }
+    }
+
+    /**
+     * Handle a job failure.
+     *
+     * @return void
+     */
+    public function failed(\Throwable $exception)
+    {
+        if (config('federation.activitypub.delivery.logger.enabled')) {
+            Log::error('NewStatusPipeline failed for status '.$this->status->id, [
+                'exception' => $exception->getMessage(),
+                '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()