From 6c13debd56d4119552113a81c716d480766f31f9 Mon Sep 17 00:00:00 2001 From: Daniel Supernault Date: Mon, 10 Feb 2025 19:17:39 -0700 Subject: [PATCH 1/2] Update NewStatusPipeline, replaces #5706 --- app/Jobs/StatusPipeline/NewStatusPipeline.php | 104 +++++++++++++++++- 1 file changed, 99 insertions(+), 5 deletions(-) diff --git a/app/Jobs/StatusPipeline/NewStatusPipeline.php b/app/Jobs/StatusPipeline/NewStatusPipeline.php index 8ccb2926f..79a238836 100644 --- a/app/Jobs/StatusPipeline/NewStatusPipeline.php +++ b/app/Jobs/StatusPipeline/NewStatusPipeline.php @@ -2,6 +2,7 @@ namespace App\Jobs\StatusPipeline; +use App\Media; use App\Status; use Cache; use Illuminate\Bus\Queueable; @@ -9,6 +10,7 @@ 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 @@ -16,7 +18,7 @@ class NewStatusPipeline implements ShouldQueue use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; protected $status; - + /** * Delete the job if its models no longer exist. * @@ -24,9 +26,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 +64,80 @@ class NewStatusPipeline implements ShouldQueue */ public function handle() { - StatusEntityLexer::dispatch($this->status); + // 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')) { + $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 && 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.'); + } + $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. + * + * @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, [ + 'exception' => $exception->getMessage(), + 'trace' => $exception->getTraceAsString() + ]); + } } } From 863ad8e107f42dceb8864ff8c3f5053b6e057426 Mon Sep 17 00:00:00 2001 From: Daniel Supernault Date: Mon, 10 Feb 2025 20:07:33 -0700 Subject: [PATCH 2/2] Update NewStatusPipeline, improve fallback --- app/Jobs/StatusPipeline/NewStatusPipeline.php | 32 +++++++------------ app/Media.php | 3 +- 2 files changed, 14 insertions(+), 21 deletions(-) diff --git a/app/Jobs/StatusPipeline/NewStatusPipeline.php b/app/Jobs/StatusPipeline/NewStatusPipeline.php index 79a238836..2c825cbc0 100644 --- a/app/Jobs/StatusPipeline/NewStatusPipeline.php +++ b/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(), ]); } } diff --git a/app/Media.php b/app/Media.php index 9ecc7b17e..51bc21152 100644 --- a/app/Media.php +++ b/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()