diff --git a/app/Http/Controllers/Api/ApiV1Controller.php b/app/Http/Controllers/Api/ApiV1Controller.php index 39eab3005..c6a278120 100644 --- a/app/Http/Controllers/Api/ApiV1Controller.php +++ b/app/Http/Controllers/Api/ApiV1Controller.php @@ -2258,7 +2258,20 @@ class ApiV1Controller extends Controller $resource = new Fractal\Resource\Item($media, new MediaTransformer); $res = $this->fractal->createData($resource)->toArray(); - return $this->json($res); + $processing = (bool) config_cache('pixelfed.cloud_storage') + && ! config('pixelfed.media_fast_process') + && in_array($media->mime, [ + 'image/jpg', + 'image/jpeg', + 'image/png', + 'image/webp', + 'image/heic', + 'image/avif', + 'video/mp4', + ]) + && ! $media->cdn_url; + + return $this->json($res, $processing ? 206 : 200); } /** diff --git a/app/Jobs/StatusPipeline/NewStatusPipeline.php b/app/Jobs/StatusPipeline/NewStatusPipeline.php index 5fd7b0449..a08bb6288 100644 --- a/app/Jobs/StatusPipeline/NewStatusPipeline.php +++ b/app/Jobs/StatusPipeline/NewStatusPipeline.php @@ -9,14 +9,36 @@ use Illuminate\Contracts\Queue\ShouldQueue; use Illuminate\Foundation\Bus\Dispatchable; use Illuminate\Queue\InteractsWithQueue; use Illuminate\Queue\SerializesModels; +use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\Log; class NewStatusPipeline implements ShouldQueue { use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; + /** + * Seconds between checks while attached media is still being processed. + */ + public const MEDIA_WAIT_SECONDS = 10; + + /** + * How many checks to make before giving up (30 x 10s = 5 minutes). + */ + public const MEDIA_WAIT_MAX = 30; + + /** + * Seconds to hold the publish lock. Long enough to cover the waiting + * job and the media pipeline both dispatching for the same status. + */ + private const LOCK_TTL = 900; + protected $status; + /** + * How many times this status has already been re-queued waiting on media. + */ + protected int $mediaWait; + /** * Delete the job if its models no longer exist. * @@ -42,9 +64,10 @@ class NewStatusPipeline implements ShouldQueue * * @return void */ - public function __construct(Status $status) + public function __construct(Status $status, int $mediaWait = 0) { $this->status = $status; + $this->mediaWait = $mediaWait; } /** @@ -68,22 +91,54 @@ class NewStatusPipeline implements ShouldQueue // 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($this->status->id) + $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. Instead, do nothing here. The media pipeline will re-call the NewStatusPipeline - // once all media items are finished processing + // 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; } diff --git a/app/Services/MediaStorageService.php b/app/Services/MediaStorageService.php index ddcebb3d3..bb7f8a739 100644 --- a/app/Services/MediaStorageService.php +++ b/app/Services/MediaStorageService.php @@ -31,7 +31,6 @@ class MediaStorageService if ((bool) config_cache('pixelfed.cloud_storage') == true && config('filesystems.default') === 'local') { return (new self)->cloudMove($media); } - } public static function avatar($avatar, $local = false, $skipRecentCheck = false) @@ -57,16 +56,40 @@ class MediaStorageService (new self)->localToCloud($media); } - if ($media->status_id && config_cache('pixelfed.cloud_storage') && ! config('pixelfed.media_fast_process')) { - $still_processing = Media::whereStatusId($media->status_id) + /* + * Read status_id fresh from the database. + * + * $media was unserialized when MediaStoragePipeline started. A status + * can be attached to it (POST /api/v1/statuses) while the upload above + * is in flight, in which case $media->status_id is a stale null and + * the NewStatusPipeline dispatched by the controller has already + * returned early because cdn_url was not set yet. Trusting the stale + * value here means that post is never lexed or federated. + */ + $statusId = Media::whereKey($media->id)->value('status_id'); + + if (! $statusId) { + return; + } + + if ($statusId != $media->status_id) { + // Attached mid-upload: localToCloud() skipped these with the stale null. + Cache::forget('pf:status:ap:v1:sid:'.$statusId); + Cache::forget('status:transformer:media:attachments:'.$statusId); + MediaService::del($statusId); + StatusService::del($statusId, false); + } + + if (config_cache('pixelfed.cloud_storage') && ! config('pixelfed.media_fast_process')) { + $still_processing = Media::whereStatusId($statusId) ->whereNull('cdn_url') ->exists(); if (! $still_processing) { // In this configuration, publishing the status is delayed until the media uploads // Since all media have been processed, we can kick the NewStatusPipeline job // N.B. there's a timing condition with multiple MediaStorageService workers matching this if statement - // However, it's acceptable to publish the same status multiple times to ActivityPub - $status = Status::where('id', $media->status_id)->first(); // This could be null if the status was deleted + // NewStatusPipeline holds a short lock so the status is only lexed and federated once + $status = Status::where('id', $statusId)->first(); // This could be null if the status was deleted if ($status) { NewStatusPipeline::dispatch($status); } diff --git a/app/Util/ActivityPub/Helpers.php b/app/Util/ActivityPub/Helpers.php index 016ac4a45..bc36633a5 100644 --- a/app/Util/ActivityPub/Helpers.php +++ b/app/Util/ActivityPub/Helpers.php @@ -247,6 +247,10 @@ class Helpers */ public static function isValidUri(Uri $uri): bool { + if (! $uri) { + return false; + } + if (strtolower($uri->getScheme()) !== 'https') { return false; }