*/ public $backoff = [5, 10]; /** * Create a new job instance. * * @return void */ public function __construct(Status $status, int $mediaWait = 0) { $this->status = $status; $this->mediaWait = $mediaWait; } /** * Execute the job. * * @return void */ public function handle() { $status = $this->status; // Verify status exists if (! $status) { Log::info('NewStatusPipeline: Status no longer exists, skipping job'); return; } if (! Status::where('id', $status->id)->exists()) { // The status has already been deleted by the time the job is running // Don't publish the status, and just no-op return; } if (config_cache('pixelfed.cloud_storage') && ! config('pixelfed.media_fast_process')) { $still_processing = Media::whereStatusId($status->id) ->whereNull('cdn_url') ->exists(); if ($still_processing) { // The media items in the status are still being processed. // We can't publish the status to ActivityPub because the final remote URL is not // yet known. // // MediaStorageService::cloudStore() re-dispatches this job when the last attachment // lands on cloud storage, but that only helps if the media pipeline actually gets // there. Poll as a fallback so a post is never silently left unpublished. if ($this->mediaWait >= self::MEDIA_WAIT_MAX) { Log::warning( "NewStatusPipeline: Media for status {$status->id} never finished processing, giving up after " .(self::MEDIA_WAIT_MAX * self::MEDIA_WAIT_SECONDS).'s' ); return; } self::dispatch($status, $this->mediaWait + 1) ->onQueue($this->queue) ->delay(now()->addSeconds(self::MEDIA_WAIT_SECONDS)); return; } } /* * Both the fallback above and MediaStorageService::cloudStore() can * dispatch this job for the same status once its media is ready. * Only the first one through should lex and federate it. */ $lock = 'pf:status:new-pipeline:'.$status->id; if (! Cache::add($lock, 1, self::LOCK_TTL)) { return; } try { StatusEntityLexer::dispatch($status); } catch (\Exception $e) { // Let the queue retry take the lock next time. Cache::forget($lock); Log::warning("NewStatusPipeline: Failed to dispatch StatusEntityLexer for status {$status->id}: ".$e->getMessage()); throw $e; } } }