From 56b96359c3f73ba861311a55081c4ab3ef914829 Mon Sep 17 00:00:00 2001 From: Daniel Supernault Date: Mon, 21 Sep 2026 00:35:27 -0600 Subject: [PATCH 1/3] Fix fetching by properly retrying certain error responses --- .../Concerns/RetriesWhenActorUnavailable.php | 67 +++ app/Jobs/InboxPipeline/InboxValidator.php | 10 +- app/Jobs/InboxPipeline/InboxWorker.php | 10 +- .../DeliverQuoteActivityPipeline.php | 126 ++++++ .../RevokeQuoteAuthorizationPipeline.php | 37 -- app/Services/ActivityPubDeliveryService.php | 23 +- app/Services/ActivityPubFetchService.php | 78 +++- app/Services/QuoteService.php | 25 +- app/Util/ActivityPub/Helpers.php | 48 +++ .../Federation/InboxFirstContactTest.php | 403 ++++++++++++++++++ tests/Feature/Federation/QuotePostsTest.php | 113 ++++- 11 files changed, 852 insertions(+), 88 deletions(-) create mode 100644 app/Jobs/InboxPipeline/Concerns/RetriesWhenActorUnavailable.php create mode 100644 app/Jobs/QuotePipeline/DeliverQuoteActivityPipeline.php delete mode 100644 app/Jobs/QuotePipeline/RevokeQuoteAuthorizationPipeline.php create mode 100644 tests/Feature/Federation/InboxFirstContactTest.php diff --git a/app/Jobs/InboxPipeline/Concerns/RetriesWhenActorUnavailable.php b/app/Jobs/InboxPipeline/Concerns/RetriesWhenActorUnavailable.php new file mode 100644 index 000000000..529a19e10 --- /dev/null +++ b/app/Jobs/InboxPipeline/Concerns/RetriesWhenActorUnavailable.php @@ -0,0 +1,67 @@ + + */ + public const ACTOR_RETRY_DELAYS = [180, 900, 3600]; + + /** + * Set by verifySignature() when the signing actor is unknown and could + * not be fetched for a reason that looks temporary. + */ + protected bool $actorUnavailable = false; + + protected function markActorUnavailable(mixed $actorUrl): void + { + $this->actorUnavailable = is_string($actorUrl) + && Helpers::fetchFailedTransiently($actorUrl); + } + + /** + * Release the job for another attempt if the actor was temporarily + * unreachable and there are attempts left. Returns true when released. + */ + protected function retryLaterIfActorUnavailable(): bool + { + if (! $this->actorUnavailable) { + return false; + } + + $attempt = $this->attempts(); + + if ($attempt > count(self::ACTOR_RETRY_DELAYS)) { + return false; + } + + $this->release(self::ACTOR_RETRY_DELAYS[$attempt - 1]); + + return true; + } +} diff --git a/app/Jobs/InboxPipeline/InboxValidator.php b/app/Jobs/InboxPipeline/InboxValidator.php index c6e8f5f15..0502899d7 100644 --- a/app/Jobs/InboxPipeline/InboxValidator.php +++ b/app/Jobs/InboxPipeline/InboxValidator.php @@ -2,6 +2,7 @@ namespace App\Jobs\InboxPipeline; +use App\Jobs\InboxPipeline\Concerns\RetriesWhenActorUnavailable; use App\Models\Profile; use App\Services\FollowersSyncService; use App\Util\ActivityPub\Helpers; @@ -17,6 +18,7 @@ use Illuminate\Support\Lottery; class InboxValidator implements ShouldQueue { use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; + use RetriesWhenActorUnavailable; protected $username; @@ -26,7 +28,9 @@ class InboxValidator implements ShouldQueue public $timeout = 300; - public $tries = 1; + // One attempt plus the retries in RetriesWhenActorUnavailable. Exceptions + // still fail the job immediately because of $maxExceptions below. + public $tries = 4; public $maxExceptions = 1; @@ -89,6 +93,8 @@ class InboxValidator implements ShouldQueue return; } + + $this->retryLaterIfActorUnavailable(); } protected function verifySignature($headers, $profile, $payload) @@ -158,6 +164,8 @@ class InboxValidator implements ShouldQueue $actor = Helpers::profileFirstOrNew($claimedActor); } if (! $actor) { + $this->markActorUnavailable($claimedActor); + return false; } // Rebind: the profile resolved by keyId must belong to the keyId host. diff --git a/app/Jobs/InboxPipeline/InboxWorker.php b/app/Jobs/InboxPipeline/InboxWorker.php index adc893ca5..8f3a80a23 100644 --- a/app/Jobs/InboxPipeline/InboxWorker.php +++ b/app/Jobs/InboxPipeline/InboxWorker.php @@ -2,6 +2,7 @@ namespace App\Jobs\InboxPipeline; +use App\Jobs\InboxPipeline\Concerns\RetriesWhenActorUnavailable; use App\Models\Profile; use App\Services\FollowersSyncService; use App\Util\ActivityPub\Helpers; @@ -16,6 +17,7 @@ use Illuminate\Support\Facades\Cache; class InboxWorker implements ShouldQueue { use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; + use RetriesWhenActorUnavailable; protected $headers; @@ -23,7 +25,9 @@ class InboxWorker implements ShouldQueue public $timeout = 300; - public $tries = 1; + // One attempt plus the retries in RetriesWhenActorUnavailable. Exceptions + // still fail the job immediately because of $maxExceptions below. + public $tries = 4; public $maxExceptions = 1; @@ -70,6 +74,8 @@ class InboxWorker implements ShouldQueue return; } + + $this->retryLaterIfActorUnavailable(); } protected function verifySignature($headers, $payload) @@ -141,6 +147,8 @@ class InboxWorker implements ShouldQueue $signer = Helpers::profileFirstOrNew($claimedActor); } if (! $signer) { + $this->markActorUnavailable($claimedActor); + return false; } diff --git a/app/Jobs/QuotePipeline/DeliverQuoteActivityPipeline.php b/app/Jobs/QuotePipeline/DeliverQuoteActivityPipeline.php new file mode 100644 index 000000000..508318f61 --- /dev/null +++ b/app/Jobs/QuotePipeline/DeliverQuoteActivityPipeline.php @@ -0,0 +1,126 @@ + + */ + public const RETRY_DELAYS = [30, 120, 600, 3600]; + + /** + * Statuses below 500 that are still worth another attempt. + */ + private const array RETRYABLE_STATUSES = [408, 425, 429]; + + public $timeout = 60; + + // One attempt plus the retries above + public $tries = 5; + + public $maxExceptions = 1; + + /** + * @param array $activity + */ + public function __construct( + protected int $fromProfileId, + protected int $toProfileId, + protected array $activity + ) {} + + /** + * @return array + */ + public function activity(): array + { + return $this->activity; + } + + public function handle(): void + { + $from = Profile::find($this->fromProfileId); + $to = Profile::find($this->toProfileId); + + if (! $from || ! $to || $from->domain !== null || $from->status !== null) { + return; + } + + $inbox = $to->sharedInbox ?? $to->inbox_url; + + if (! $inbox) { + Log::info('DeliverQuoteActivityPipeline: remote actor has no inbox', [ + 'profile_id' => $from->id, + 'actor_id' => $to->id, + ]); + + return; + } + + if (! app()->environment('production')) { + return; + } + + try { + $response = ActivityPubDeliveryService::queue() + ->from($from) + ->to($inbox) + ->payload($this->activity) + ->deliver(); + } catch (InvalidDeliveryDestinationException|InvalidArgumentException $e) { + // Banned host, bad inbox URL, sender without keys: retrying cannot help + return; + } + + // Null means nothing reached the remote (connection failure, or the + // host is currently marked unavailable), which is worth retrying. + if ($response && ! self::shouldRetry($response->status())) { + return; + } + + $attempt = $this->attempts(); + + if ($attempt > count(self::RETRY_DELAYS)) { + Log::warning('DeliverQuoteActivityPipeline: giving up', [ + 'profile_id' => $from->id, + 'actor_id' => $to->id, + 'type' => $this->activity['type'] ?? null, + 'id' => $this->activity['id'] ?? null, + 'status' => $response?->status(), + ]); + + return; + } + + $this->release(self::RETRY_DELAYS[$attempt - 1]); + } + + private static function shouldRetry(int $status): bool + { + return $status >= 500 || in_array($status, self::RETRYABLE_STATUSES, true); + } +} diff --git a/app/Jobs/QuotePipeline/RevokeQuoteAuthorizationPipeline.php b/app/Jobs/QuotePipeline/RevokeQuoteAuthorizationPipeline.php deleted file mode 100644 index e360b6f16..000000000 --- a/app/Jobs/QuotePipeline/RevokeQuoteAuthorizationPipeline.php +++ /dev/null @@ -1,37 +0,0 @@ -find($this->authorizationId); - - if (! $auth || ! $auth->isRevoked()) { - return; - } - - QuoteService::sendDelete($auth); - } -} diff --git a/app/Services/ActivityPubDeliveryService.php b/app/Services/ActivityPubDeliveryService.php index d1d445591..354b92c99 100644 --- a/app/Services/ActivityPubDeliveryService.php +++ b/app/Services/ActivityPubDeliveryService.php @@ -66,12 +66,25 @@ class ActivityPubDeliveryService $this->queueDelivery(); } + /** + * Same as send(), but hands back the remote's response so the caller + * can decide whether to try again. Null means nothing reached the + * remote: delivery is skipped outside production, the host is marked + * unavailable, or the connection failed. + * + * @throws InvalidDeliveryDestinationException when the inbox URL fails validation + */ + public function deliver(): ?Response + { + return $this->queueDelivery(); + } + /** * Deliver a single ActivityPub activity. * * @throws InvalidDeliveryDestinationException when the inbox URL fails validation */ - protected function queueDelivery(): void + protected function queueDelivery(): ?Response { if (! $this->sender) { throw new InvalidArgumentException('Missing ActivityPub sender.'); @@ -110,7 +123,7 @@ class ActivityPubDeliveryService 'url' => $url, ]); - return; + return null; } if ($domain && DeliveryHostService::isUnavailable($domain)) { @@ -119,7 +132,7 @@ class ActivityPubDeliveryService 'url' => $url, ]); - return; + return null; } try { @@ -152,6 +165,8 @@ class ActivityPubDeliveryService $response ); } + + return $response; } catch (Throwable $e) { if ($domain && $e instanceof ConnectionException) { DeliveryHostService::recordFailure($domain); @@ -172,7 +187,7 @@ class ActivityPubDeliveryService // Other exception types (invalid sender/destination, signing, // serialization) still throw, as they did before the rewrite. if ($e instanceof ConnectionException) { - return; + return null; } throw $e; diff --git a/app/Services/ActivityPubFetchService.php b/app/Services/ActivityPubFetchService.php index 3513ea439..713bf8838 100644 --- a/app/Services/ActivityPubFetchService.php +++ b/app/Services/ActivityPubFetchService.php @@ -19,17 +19,62 @@ class ActivityPubFetchService private const MAX_RESPONSE_SIZE = 2 * 1024 * 1024; + /** + * Statuses worth another attempt later. Everything else in the 4xx + * range (401, 403, 404, 410...) is the remote telling us no. + */ + private const array TRANSIENT_STATUSES = [408, 425, 429]; + + /** + * Whether the most recent failed fetch in this process looked temporary + * (timeout, connection error, DNS, 5xx, 429) rather than permanent. + * Null when the last fetch did not fail. + */ + private static ?bool $lastFailureTransient = null; + + /** + * True when the fetch that just returned nothing failed in a way that + * is likely to work if tried again later. Only meaningful right after + * a call to get() or fetchRequest() came back empty. + */ + public static function lastFailureWasTransient(): bool + { + return self::$lastFailureTransient === true; + } + + private static function isTransientStatus(int $status): bool + { + return $status >= 500 || in_array($status, self::TRANSIENT_STATUSES, true); + } + + /** + * Record why a fetch failed. Always returns null so call sites can + * `return self::failed(...)`. + */ + private static function failed(bool $transient): null + { + self::$lastFailureTransient = $transient; + + return null; + } + public static function get($url, $validateUrl = true) { + self::$lastFailureTransient = null; + $url = Helpers::validateUrl($url); if (! $url) { + self::failed(false); + return false; } $host = parse_url($url, PHP_URL_HOST); if (! $host) { + self::failed(false); + return false; } @@ -49,26 +94,30 @@ class ActivityPubFetchService public static function fetchRequest($url, $returnJsonFormat = false) { + self::$lastFailureTransient = null; + $currentUrl = $url; for ($redirects = 0; $redirects <= self::MAX_REDIRECTS; $redirects++) { $currentUrl = Helpers::validateUrl($currentUrl); if (! $currentUrl) { - return; + return self::failed(false); } $host = parse_url($currentUrl, PHP_URL_HOST); $port = parse_url($currentUrl, PHP_URL_PORT) ?: 443; if (! $host) { - return; + return self::failed(false); } $ips = Helpers::resolvePublicIps($host); if ($ips === []) { - return; + // A host that just delivered to us but does not resolve is + // far more likely a DNS blip than a dead domain. + return self::failed(true); } $headers = self::signedHeaders($currentUrl); @@ -109,25 +158,30 @@ class ActivityPubFetchService ->connectTimeout(5) ->retry(2, 250) ->get($currentUrl); - } catch (RequestException|ConnectionException|\Throwable $e) { - return; + } catch (RequestException $e) { + // retry() throws once its attempts are used up + return self::failed(self::isTransientStatus($e->response->status())); + } catch (ConnectionException $e) { + return self::failed(true); + } catch (\Throwable $e) { + return self::failed(false); } if (in_array($res->status(), [301, 302, 303, 307, 308], true)) { if ($redirects >= self::MAX_REDIRECTS) { - return; + return self::failed(false); } $location = $res->header('Location'); if (! $location) { - return; + return self::failed(false); } $nextUrl = self::resolveRedirect($currentUrl, $location); if (! $nextUrl) { - return; + return self::failed(false); } $currentUrl = $nextUrl; @@ -136,11 +190,11 @@ class ActivityPubFetchService } if (! $res->ok()) { - return; + return self::failed(self::isTransientStatus($res->status())); } if (! self::hasValidContentType($res)) { - return; + return self::failed(false); } $body = $res->body(); @@ -149,7 +203,7 @@ class ActivityPubFetchService $body === '' || strlen($body) > self::MAX_RESPONSE_SIZE ) { - return; + return self::failed(false); } if (! $returnJsonFormat) { @@ -164,7 +218,7 @@ class ActivityPubFetchService JSON_THROW_ON_ERROR ); } catch (\JsonException) { - return; + return self::failed(false); } } diff --git a/app/Services/QuoteService.php b/app/Services/QuoteService.php index 3aafbd7e7..78397fbcf 100644 --- a/app/Services/QuoteService.php +++ b/app/Services/QuoteService.php @@ -2,13 +2,11 @@ namespace App\Services; -use App\Jobs\QuotePipeline\RevokeQuoteAuthorizationPipeline; +use App\Jobs\QuotePipeline\DeliverQuoteActivityPipeline; use App\Models\Profile; use App\Models\QuoteAuthorization; use App\Models\Status; -use App\Util\ActivityPub\Helpers; use Illuminate\Support\Facades\Cache; -use Illuminate\Support\Facades\Log; /** * FEP-044f: Consent-respecting quote posts. @@ -438,7 +436,7 @@ class QuoteService /** * Withdraw consent. The row is kept (state revoked) so a repeat request * for the same quote post is auto-rejected, and the stamp URL serves - * 410 Gone. The Delete activity is sent from a queued job. + * 410 Gone. The Delete is delivered from a queued job, with retries. */ public static function revoke(QuoteAuthorization $auth, bool $notify = true): void { @@ -451,7 +449,7 @@ class QuoteService $auth->save(); if ($notify) { - RevokeQuoteAuthorizationPipeline::dispatch($auth->id)->onQueue('high'); + self::sendDelete($auth); } } @@ -613,22 +611,15 @@ class QuoteService } /** + * Hand the activity to a queued job that retries on temporary failures. + * The quoting server sends its QuoteRequest once, so a lost Accept would + * leave the quote pending on their side for good. + * * @param array $activity */ private static function deliver(Profile $from, Profile $to, array $activity): void { - $inbox = $to->sharedInbox ?? $to->inbox_url; - - if (! $inbox) { - Log::info('QuoteService: remote actor has no inbox', [ - 'profile_id' => $from->id, - 'actor_id' => $to->id, - ]); - - return; - } - - Helpers::sendSignedObject($from, $inbox, $activity); + DeliverQuoteActivityPipeline::dispatch($from->id, $to->id, $activity)->onQueue('high'); } /** diff --git a/app/Util/ActivityPub/Helpers.php b/app/Util/ActivityPub/Helpers.php index e07360840..ced7124e2 100644 --- a/app/Util/ActivityPub/Helpers.php +++ b/app/Util/ActivityPub/Helpers.php @@ -864,9 +864,57 @@ class Helpers : now()->addMinutes(self::FETCH_CACHE_TTL) ); + /* + * Alongside a cached failure, remember whether it looked temporary + * (timeout, connection error, 5xx, 429) as opposed to the remote + * saying no (401, 403, 404, 410, not ActivityPub). Callers that can + * retry later use fetchFailedTransiently() to tell the two apart. + * The marker lives exactly as long as the cached failure. + */ + $marker = self::fetchTransientKey($url); + + if ($res === false && ActivityPubFetchService::lastFailureWasTransient()) { + Cache::put($marker, 1, self::FETCH_NEGATIVE_TTL); + } else { + Cache::forget($marker); + } + return $res; } + private static function fetchTransientKey(string $url): string + { + return 'helpers:url:fetcher:transient:sha256-'.hash('sha256', $url); + } + + /** + * Did the last attempt to fetch this URL fail in a way that is worth + * retrying later? Only answers while that failure is still negatively + * cached, so a retry has to wait longer than FETCH_NEGATIVE_TTL or it + * will just find the cached failure again. + */ + public static function fetchFailedTransiently(mixed $url): bool + { + if (! is_string($url) || $url === '') { + return false; + } + + $candidates = [$url]; + $validated = self::validateUrl($url); + + if (is_string($validated)) { + $candidates[] = $validated; + } + + foreach (array_unique($candidates) as $candidate) { + if (Cache::has(self::fetchTransientKey($candidate))) { + return true; + } + } + + return false; + } + public static function fetchCacheKey(string $url): string { return 'helpers:url:fetcher:sha256-'.hash('sha256', $url); diff --git a/tests/Feature/Federation/InboxFirstContactTest.php b/tests/Feature/Federation/InboxFirstContactTest.php new file mode 100644 index 000000000..0a895483c --- /dev/null +++ b/tests/Feature/Federation/InboxFirstContactTest.php @@ -0,0 +1,403 @@ + false, + 'federation.activitypub.enabled' => true, + ]); +}); + +function fcLocalUser(): User +{ + $user = User::factory()->create(); + $user->refresh(); + + return $user; +} + +function fcLocalStatus(Profile $profile): Status +{ + return Status::factory()->photo()->create(['profile_id' => $profile->id]); +} + +/** + * Seed the DNS and banned-domain caches so URL validation passes without a + * network lookup. Call after factories, the lazy refresh can flush the cache. + */ +function fcSeedHosts(array $hosts = ['remote.example']): void +{ + $hosts[] = config('pixelfed.domain.app'); + + foreach ($hosts as $host) { + Cache::put( + 'helpers:url:public-ips:v2:'.hash('xxh128', $host), + ['state' => Helpers::URL_OK, 'ips' => ['203.0.113.40']], + 3600 + ); + } + + Cache::put('instances:banned:domains', [], 1209600); +} + +function fcInProduction(callable $fn): mixed +{ + $app = app(); + $previous = $app['env']; + $app['env'] = 'production'; + + try { + return $fn(); + } finally { + $app['env'] = $previous; + } +} + +function fcSentOfType(string $type): array +{ + return Queue::pushed(DeliverQuoteActivityPipeline::class) + ->map(fn (DeliverQuoteActivityPipeline $job) => $job->activity()) + ->filter(fn ($activity) => $activity['type'] === $type) + ->values() + ->all(); +} + +/** + * A remote actor Pixelfed has never seen, with a real key pair so requests + * can be signed and verified end to end. + */ +function fcStranger(string $domain = 'remote.example', string $username = 'alice'): array +{ + $key = openssl_pkey_new(['private_key_bits' => 2048, 'private_key_type' => OPENSSL_KEYTYPE_RSA]); + $actor = "https://{$domain}/users/{$username}"; + + return [ + 'key' => $key, + 'actor' => $actor, + 'document' => [ + '@context' => ['https://www.w3.org/ns/activitystreams', 'https://w3id.org/security/v1'], + 'id' => $actor, + 'type' => 'Person', + 'following' => $actor.'/following', + 'followers' => $actor.'/followers', + 'inbox' => $actor.'/inbox', + 'outbox' => $actor.'/outbox', + 'preferredUsername' => $username, + 'name' => ucfirst($username), + 'summary' => '', + 'url' => "https://{$domain}/@{$username}", + 'manuallyApprovesFollowers' => false, + 'publicKey' => [ + 'id' => $actor.'#main-key', + 'owner' => $actor, + 'publicKeyPem' => openssl_pkey_get_details($key)['key'], + ], + 'endpoints' => ['sharedInbox' => "https://{$domain}/inbox"], + ], + ]; +} + +/** + * Fake the stranger's actor endpoint. $failures is a list of responses to + * serve before the real document, 'connection' simulates a network error. + */ +function fcFakeStranger(array $stranger, array $failures = []): object +{ + $state = new stdClass; + $state->fetches = 0; + + Http::fake([ + parse_url($stranger['actor'], PHP_URL_HOST).'/users/*' => function () use ($state, $stranger, &$failures) { + $state->fetches++; + + $failure = array_shift($failures); + + if ($failure === 'connection') { + throw new ConnectionException('timed out'); + } + + if (is_int($failure)) { + return Http::response('nope', $failure); + } + + // A string body, Http::response() forces application/json onto arrays + return Http::response(json_encode($stranger['document']), 200, [ + 'Content-Type' => 'application/activity+json', + ]); + }, + '*' => Http::response('', 202), + ]); + + return $state; +} + +/** + * What Mastodon sends: the quote post inlined as `instrument`, delivered + * to the quoted account's own inbox. + */ +function fcMastodonRequest(array $stranger, Status $status): string +{ + $quoteUrl = $stranger['actor'].'/statuses/115221849202938471'; + + return json_encode([ + '@context' => [ + 'https://www.w3.org/ns/activitystreams', + ['QuoteRequest' => 'https://w3id.org/fep/044f#QuoteRequest'], + ], + 'id' => $stranger['actor'].'/quote_requests/0c0f6c1e-5f0b-4c55-9a53-0d6c6f0b7a11', + 'type' => 'QuoteRequest', + 'actor' => $stranger['actor'], + 'object' => $status->url(), + 'instrument' => [ + 'id' => $quoteUrl, + 'type' => 'Note', + 'summary' => null, + 'inReplyTo' => null, + 'published' => now()->toIso8601String(), + 'url' => str_replace('/users/', '/@', $stranger['actor']).'/115221849202938471', + 'attributedTo' => $stranger['actor'], + 'to' => ['https://www.w3.org/ns/activitystreams#Public'], + 'cc' => [$stranger['actor'].'/followers'], + 'sensitive' => false, + 'content' => '

nice shot

', + 'contentMap' => ['en' => '

nice shot

'], + 'attachment' => [], + 'tag' => [], + 'replies' => [ + 'id' => $quoteUrl.'/replies', + 'type' => 'Collection', + 'first' => ['type' => 'CollectionPage', 'partOf' => $quoteUrl.'/replies', 'items' => []], + ], + 'quote' => $status->url(), + '_misskey_quote' => $status->url(), + 'quoteUri' => $status->url(), + 'interactionPolicy' => ['canQuote' => ['automaticApproval' => ['https://www.w3.org/ns/activitystreams#Public']]], + ], + ], JSON_UNESCAPED_SLASHES); +} + +function fcSignedHeaders(array $stranger, string $path, string $body): array +{ + $host = config('pixelfed.domain.app'); + $date = now()->toRfc7231String(); + $digest = 'SHA-256='.base64_encode(hash('sha256', $body, true)); + + openssl_sign( + "(request-target): post {$path}\nhost: {$host}\ndate: {$date}\ndigest: {$digest}", + $signature, + $stranger['key'], + OPENSSL_ALGO_SHA256 + ); + + return [ + 'host' => [$host], + 'date' => [$date], + 'digest' => [$digest], + 'content-type' => ['application/activity+json'], + 'signature' => ['keyId="'.$stranger['actor'].'#main-key",algorithm="rsa-sha256",headers="(request-target) host date digest",signature="'.base64_encode($signature).'"'], + ]; +} + +/** + * Run whatever the inbox job handed to ActivityHandler, the queue is faked. + */ +function fcRunHandlers(): void +{ + Queue::pushed(ActivityHandler::class)->each(fn ($job) => fcInProduction(fn () => $job->handle())); +} + +describe('first contact', function () { + it('answers a signed Mastodon request from an account it has never seen, via the user inbox', function () { + $user = fcLocalUser(); + $status = fcLocalStatus($user->profile); + $alice = fcStranger(); + fcFakeStranger($alice); + fcSeedHosts(); + + $body = fcMastodonRequest($alice, $status); + $headers = fcSignedHeaders($alice, "/users/{$user->profile->username}/inbox", $body); + + expect(Profile::whereRemoteUrl($alice['actor'])->exists())->toBeFalse(); + + fcInProduction(fn () => (new InboxValidator($user->profile->username, $headers, $body))->handle()); + fcRunHandlers(); + + $accept = fcSentOfType('Accept'); + + expect(Profile::whereRemoteUrl($alice['actor'])->exists())->toBeTrue() + ->and(QuoteAuthorization::approved()->count())->toBe(1) + ->and($accept)->toHaveCount(1) + ->and($accept[0]['object']['id'])->toBe(json_decode($body, true)['id']) + ->and($accept[0]['result'])->toBe(QuoteAuthorization::first()->permalink()); + }); + + it('answers the same request via the shared inbox', function () { + $user = fcLocalUser(); + $status = fcLocalStatus($user->profile); + $alice = fcStranger(); + fcFakeStranger($alice); + fcSeedHosts(); + + $body = fcMastodonRequest($alice, $status); + $headers = fcSignedHeaders($alice, '/f/inbox', $body); + + fcInProduction(fn () => (new InboxWorker($headers, $body))->handle()); + fcRunHandlers(); + + expect(QuoteAuthorization::approved()->count())->toBe(1) + ->and(fcSentOfType('Accept'))->toHaveCount(1); + }); + + it('releases the job instead of dropping it when the actor is temporarily unreachable', function (mixed $failure) { + $user = fcLocalUser(); + $status = fcLocalStatus($user->profile); + $alice = fcStranger(); + // retry(2) inside the fetch service means one fetch is two requests + $remote = fcFakeStranger($alice, [$failure, $failure]); + fcSeedHosts(); + + $body = fcMastodonRequest($alice, $status); + $headers = fcSignedHeaders($alice, "/users/{$user->profile->username}/inbox", $body); + + $job = (new InboxValidator($user->profile->username, $headers, $body))->withFakeQueueInteractions(); + fcInProduction(fn () => $job->handle()); + + $job->assertReleased(delay: 180); + + expect(QuoteAuthorization::count())->toBe(0) + ->and(Queue::pushed(DeliverQuoteActivityPipeline::class))->toBeEmpty(); + + // The retry runs once the cached failure has lapsed + $this->travel(Helpers::FETCH_NEGATIVE_TTL + 1)->seconds(); + + $retry = (new InboxValidator($user->profile->username, $headers, $body))->withFakeQueueInteractions(); + fcInProduction(fn () => $retry->handle()); + fcRunHandlers(); + + $retry->assertNotReleased(); + + expect($remote->fetches)->toBe(3) + ->and(QuoteAuthorization::approved()->count())->toBe(1) + ->and(fcSentOfType('Accept'))->toHaveCount(1); + })->with([ + '503' => [503], + '429' => [429], + 'connection error' => ['connection'], + ]); + + it('also releases from the shared inbox', function () { + $user = fcLocalUser(); + $status = fcLocalStatus($user->profile); + $alice = fcStranger(); + fcFakeStranger($alice, [503, 503]); + fcSeedHosts(); + + $body = fcMastodonRequest($alice, $status); + + $job = (new InboxWorker(fcSignedHeaders($alice, '/f/inbox', $body), $body))->withFakeQueueInteractions(); + fcInProduction(fn () => $job->handle()); + + $job->assertReleased(delay: 180); + }); + + it('still drops at once when the actor is a definite no', function (int $status) { + $user = fcLocalUser(); + $post = fcLocalStatus($user->profile); + $alice = fcStranger(); + $remote = fcFakeStranger($alice, [$status, $status, $status, $status]); + fcSeedHosts(); + + $body = fcMastodonRequest($alice, $post); + $headers = fcSignedHeaders($alice, "/users/{$user->profile->username}/inbox", $body); + + $job = (new InboxValidator($user->profile->username, $headers, $body))->withFakeQueueInteractions(); + fcInProduction(fn () => $job->handle()); + + $job->assertNotReleased(); + + $fetches = $remote->fetches; + + // A burst from the same actor is absorbed by the cached failure + $again = (new InboxValidator($user->profile->username, $headers, $body))->withFakeQueueInteractions(); + fcInProduction(fn () => $again->handle()); + + $again->assertNotReleased(); + + expect($remote->fetches)->toBe($fetches); + })->with([401, 403, 404, 410]); + + it('does not release for a bad signature', function () { + $user = fcLocalUser(); + $status = fcLocalStatus($user->profile); + $alice = fcStranger(); + $mallory = fcStranger('remote.example', 'mallory'); + fcFakeStranger($alice); + fcSeedHosts(); + + $body = fcMastodonRequest($alice, $status); + $headers = fcSignedHeaders($alice, "/users/{$user->profile->username}/inbox", $body); + // Signed with someone else's key + $forged = fcSignedHeaders(['key' => $mallory['key'], 'actor' => $alice['actor']], "/users/{$user->profile->username}/inbox", $body); + + $job = (new InboxValidator($user->profile->username, $forged, $body))->withFakeQueueInteractions(); + fcInProduction(fn () => $job->handle()); + + $job->assertNotReleased(); + + expect(Queue::pushed(ActivityHandler::class))->toBeEmpty() + ->and($headers)->not->toBe($forged); + }); + + it('stops retrying once the delays are used up', function () { + $user = fcLocalUser(); + $status = fcLocalStatus($user->profile); + $alice = fcStranger(); + fcFakeStranger($alice, array_fill(0, 20, 503)); + fcSeedHosts(); + + $body = fcMastodonRequest($alice, $status); + $headers = fcSignedHeaders($alice, "/users/{$user->profile->username}/inbox", $body); + + expect((new InboxValidator($user->profile->username, $headers, $body))->tries) + ->toBe(count(InboxValidator::ACTOR_RETRY_DELAYS) + 1) + ->and((new InboxWorker($headers, $body))->tries) + ->toBe(count(InboxWorker::ACTOR_RETRY_DELAYS) + 1) + ->and(min(InboxValidator::ACTOR_RETRY_DELAYS)) + ->toBeGreaterThan(Helpers::FETCH_NEGATIVE_TTL); + }); +}); diff --git a/tests/Feature/Federation/QuotePostsTest.php b/tests/Feature/Federation/QuotePostsTest.php index 46fb310ad..d647899ec 100644 --- a/tests/Feature/Federation/QuotePostsTest.php +++ b/tests/Feature/Federation/QuotePostsTest.php @@ -1,6 +1,6 @@ Helpers::URL_OK, 'ips' => ['203.0.113.40']], + 3600 + ); } Cache::put('instances:banned:domains', [], 1209600); @@ -131,16 +138,23 @@ function quoteDeliver(Profile $actor, array $payload, ?Profile $signer = null): quoteInProduction(fn () => (new Inbox($headers, null, $payload))->handle()); } +/** + * Activities handed to the delivery job, in order. Accept, Reject and Delete + * are all delivered from DeliverQuoteActivityPipeline. + */ function quoteSentActivities(): array { - return collect(Http::recorded()) - ->filter(fn ($pair) => $pair[0]->method() === 'POST') - ->map(fn ($pair) => json_decode($pair[0]->body(), true)) - ->filter() + return Queue::pushed(DeliverQuoteActivityPipeline::class) + ->map(fn (DeliverQuoteActivityPipeline $job) => $job->activity()) ->values() ->all(); } +function quoteSentOfType(string $type): array +{ + return array_values(array_filter(quoteSentActivities(), fn ($activity) => $activity['type'] === $type)); +} + function quoteNoteObject(Status $status): array { $fractal = new Fractal\Manager; @@ -291,7 +305,13 @@ describe('QuoteRequest', function () { ->and($accept['object']['object'])->toBe($status->url()) ->and($accept['object']['instrument'])->toBe($payload['instrument']); - Http::assertSent(fn ($request) => $request->url() === 'https://remote.example/inbox'); + $job = Queue::pushed(DeliverQuoteActivityPipeline::class)->first(); + + quoteInProduction(fn () => $job->handle()); + + Http::assertSent(fn ($request) => $request->method() === 'POST' + && $request->url() === 'https://remote.example/inbox' + && json_decode($request->body(), true)['type'] === 'Accept'); }); it('rejects when the account policy is nobody', function () { @@ -536,13 +556,76 @@ describe('QuoteRequest', function () { QuoteService::revoke(QuoteAuthorization::first()); quoteDeliver($bob, quoteRequestPayload($bob, $status)); - $sent = quoteSentActivities(); - - expect($sent[1]['type'])->toBe('Reject') + expect(array_column(quoteSentActivities(), 'type'))->toBe(['Accept', 'Delete', 'Reject']) ->and(QuoteAuthorization::approved()->count())->toBe(0); }); }); +describe('delivery', function () { + function quoteDeliveryJob(): DeliverQuoteActivityPipeline + { + $user = quoteLocalUser(); + $status = quoteLocalStatus($user->profile); + $bob = quoteRemoteProfile(); + quoteSeedHosts(); + + $auth = QuoteService::authorize($status, $bob, $bob->remote_url.'/statuses/1'); + + return (new DeliverQuoteActivityPipeline( + $user->profile->id, + $bob->id, + QuoteService::acceptActivity($auth, $bob->remote_url.'/statuses/1/quote') + ))->withFakeQueueInteractions(); + } + + it('is done after a 2xx', function () { + $job = quoteDeliveryJob(); + Http::swap(new Factory); + Http::fake(['*' => Http::response('', 202)]); + + quoteInProduction(fn () => $job->handle()); + + $job->assertNotReleased(); + Http::assertSentCount(1); + }); + + it('tries again later when the remote is struggling', function (mixed $failure) { + $job = quoteDeliveryJob(); + + Http::swap(new Factory); + Http::fake(['*' => function () use ($failure) { + if ($failure === 'connection') { + throw new ConnectionException('timed out'); + } + + return Http::response('', $failure); + }]); + + quoteInProduction(fn () => $job->handle()); + + $job->assertReleased(delay: 30); + })->with([ + '500' => [500], + '503' => [503], + '429' => [429], + 'connection error' => ['connection'], + ]); + + it('gives up when the remote refuses', function (int $status) { + $job = quoteDeliveryJob(); + Http::swap(new Factory); + Http::fake(['*' => Http::response('', $status)]); + + quoteInProduction(fn () => $job->handle()); + + $job->assertNotReleased(); + })->with([400, 401, 403, 404, 410]); + + it('has one more try than it has delays', function () { + expect(quoteDeliveryJob()->tries)->toBe(count(DeliverQuoteActivityPipeline::RETRY_DELAYS) + 1); + }); +}); + describe('stamp', function () { it('serves a QuoteAuthorization that satisfies the FEP verification rules', function () { $user = quoteLocalUser(); @@ -596,11 +679,9 @@ describe('revocation', function () { QuoteService::revoke($auth); - Queue::assertPushed(RevokeQuoteAuthorizationPipeline::class, 1); - - quoteInProduction(fn () => (new RevokeQuoteAuthorizationPipeline($auth->id))->handle()); + expect(quoteSentOfType('Delete'))->toHaveCount(1); - $delete = quoteSentActivities()[0]; + $delete = quoteSentOfType('Delete')[0]; expect($delete['type'])->toBe('Delete') ->and($delete['actor'])->toBe($user->profile->permalink()) @@ -625,7 +706,7 @@ describe('revocation', function () { expect(QuoteAuthorization::approved()->count())->toBe(1) ->and((int) QuoteAuthorization::approved()->first()->actor_id)->toBe((int) $carol->id); - Queue::assertPushed(RevokeQuoteAuthorizationPipeline::class, 2); + expect(quoteSentOfType('Delete'))->toHaveCount(2); }); it('revokes every stamp issued to a domain when the author blocks it', function () { @@ -681,7 +762,7 @@ describe('settings', function () { expect($auth->fresh()->isRevoked())->toBeTrue(); - Queue::assertPushed(RevokeQuoteAuthorizationPipeline::class, 1); + expect(quoteSentOfType('Delete'))->toHaveCount(1); }); it('does not let someone revoke a stamp that is not theirs', function () { From 226785f085c54d96a351fdda6ca25b97fe5c730a Mon Sep 17 00:00:00 2001 From: Daniel Supernault Date: Mon, 21 Sep 2026 00:48:51 -0600 Subject: [PATCH 2/3] Fix RemoteReplyResolvePipeline, make unique until processing --- .../RemoteReplyResolvePipeline.php | 53 +++++++++++++++++-- 1 file changed, 48 insertions(+), 5 deletions(-) diff --git a/app/Jobs/StatusPipeline/RemoteReplyResolvePipeline.php b/app/Jobs/StatusPipeline/RemoteReplyResolvePipeline.php index 943d236a8..cd8d8f7a7 100644 --- a/app/Jobs/StatusPipeline/RemoteReplyResolvePipeline.php +++ b/app/Jobs/StatusPipeline/RemoteReplyResolvePipeline.php @@ -5,6 +5,7 @@ namespace App\Jobs\StatusPipeline; use App\Models\Profile; use App\Util\ActivityPub\Helpers; use Illuminate\Bus\Queueable; +use Illuminate\Contracts\Queue\ShouldBeUniqueUntilProcessing; use Illuminate\Contracts\Queue\ShouldQueue; use Illuminate\Foundation\Bus\Dispatchable; use Illuminate\Queue\InteractsWithQueue; @@ -23,8 +24,15 @@ use Illuminate\Support\Facades\Log; * Attempts are tracked on the job and re-dispatched with a delay instead of * using release()/tries, so an unreachable parent does not end up in * failed_jobs. + * + * Two guards keep this to one queued job per reply. The pending key spans + * the whole chain and stops a redelivered activity from parking a second + * one. The unique lock covers each queued attempt, whatever dispatched it. + * It has to be ShouldBeUniqueUntilProcessing: plain ShouldBeUnique holds the + * lock until handle() returns, so the re-dispatch in retryOrDrop() would + * fail to acquire it and be discarded without an error. */ -class RemoteReplyResolvePipeline implements ShouldQueue +class RemoteReplyResolvePipeline implements ShouldBeUniqueUntilProcessing, ShouldQueue { use Dispatchable, InteractsWithQueue, Queueable; @@ -36,10 +44,13 @@ class RemoteReplyResolvePipeline implements ShouldQueue public const array BACKOFF = [180, 900, 3600, 21600]; /** - * How long a reply is considered pending, a little over the sum of - * BACKOFF. Stops a redelivered activity from starting a second chain. + * Added to the delay of the attempt being scheduled to get the TTL of + * the pending key and the unique lock. Has to cover how far the low + * queue can run behind, plus one attempt's $timeout. If the queue lags + * more than this, both expire before the attempt runs and a redelivered + * activity can start a second chain. */ - private const int PENDING_TTL = 28800; + private const int LAG_MARGIN = 3600; public $timeout = 300; @@ -71,7 +82,7 @@ class RemoteReplyResolvePipeline implements ShouldQueue return false; } - if (! Cache::add(self::pendingKey($id), 1, self::PENDING_TTL)) { + if (! Cache::add(self::pendingKey($id), 1, self::ttlFor(0))) { return false; } @@ -87,6 +98,33 @@ class RemoteReplyResolvePipeline implements ShouldQueue return 'pf:ap:reply-resolve:pending:'.hash('sha256', $id); } + /** + * Seconds the pending key and the unique lock are held while waiting for + * the given attempt. + */ + private static function ttlFor(int $attempt): int + { + return (self::BACKOFF[$attempt] ?? max(self::BACKOFF)) + self::LAG_MARGIN; + } + + /** + * One queued attempt per reply. The lock is taken on dispatch and + * released as the attempt starts processing. + */ + public function uniqueId(): string + { + return hash('sha256', (string) Helpers::pluckval($this->object['id'] ?? null)); + } + + /** + * Safety expiry for the lock, in case the queued attempt is lost (queue + * cleared, payload evicted) and never gets to release it. + */ + public function uniqueFor(): int + { + return self::ttlFor($this->attempt); + } + public function handle(): void { $id = Helpers::pluckval($this->object['id'] ?? null); @@ -142,6 +180,11 @@ class RemoteReplyResolvePipeline implements ShouldQueue return; } + // Re-arm the pending key for the wait ahead. It is sized per attempt + // instead of once for the whole chain, so a chain that runs late + // cannot outlive it. + Cache::put(self::pendingKey($id), 1, self::ttlFor($next)); + self::dispatch($this->object, $this->profileId, $next) ->delay(now()->addSeconds(self::BACKOFF[$next])) ->onQueue('low'); From 809ec1f0e5521a05b2dd1495a409647ba38d1979 Mon Sep 17 00:00:00 2001 From: Daniel Supernault Date: Mon, 21 Sep 2026 00:52:46 -0600 Subject: [PATCH 3/3] Fix FollowServiceWarmCache pipelines --- .../FollowPipeline/FollowServiceWarmCache.php | 222 +++++++++++++----- ...lowServiceWarmCacheLargeIngestPipeline.php | 174 +++++++++++--- 2 files changed, 302 insertions(+), 94 deletions(-) diff --git a/app/Jobs/FollowPipeline/FollowServiceWarmCache.php b/app/Jobs/FollowPipeline/FollowServiceWarmCache.php index 4663876db..64e5270e3 100644 --- a/app/Jobs/FollowPipeline/FollowServiceWarmCache.php +++ b/app/Jobs/FollowPipeline/FollowServiceWarmCache.php @@ -2,7 +2,6 @@ namespace App\Jobs\FollowPipeline; -use App\Models\Follower; use App\Models\Profile; use App\Services\AccountService; use App\Services\FollowerService; @@ -14,17 +13,23 @@ use Illuminate\Queue\Middleware\WithoutOverlapping; use Illuminate\Queue\SerializesModels; use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\DB; -use Illuminate\Support\Facades\Storage; +use Throwable; class FollowServiceWarmCache implements ShouldQueue { use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; + private const INLINE_LIMIT = 100; + + private const SYNC_TTL = 604800; // 7 days + + private const PROCESSING_TTL = 21600; // 6 hours + public $profileId; public $tries = 5; - public $timeout = 5000; + public $timeout = 300; public $failOnTimeout = false; @@ -35,13 +40,15 @@ class FollowServiceWarmCache implements ShouldQueue */ public function middleware(): array { - return [(new WithoutOverlapping($this->profileId))->dontRelease()]; + return [ + (new WithoutOverlapping('follow-warm-cache:'.$this->profileId)) + ->expireAfter($this->timeout + 60) + ->dontRelease(), + ]; } /** * Create a new job instance. - * - * @return void */ public function __construct($profileId) { @@ -50,83 +57,186 @@ class FollowServiceWarmCache implements ShouldQueue /** * Execute the job. - * - * @return void */ - public function handle() + public function handle(): void { - $id = $this->profileId; + $id = (int) $this->profileId; - if (Cache::has(FollowerService::FOLLOWERS_SYNC_KEY.$id) && Cache::has(FollowerService::FOLLOWING_SYNC_KEY.$id)) { + if ($id <= 0) { + return; + } + + $followersReady = $this->isReady($id, 'followers'); + $followingReady = $this->isReady($id, 'following'); + + if ($followersReady && $followingReady) { return; } $account = AccountService::get($id, true); if (! $account) { - Cache::put(FollowerService::FOLLOWERS_SYNC_KEY.$id, 1, 604800); - Cache::put(FollowerService::FOLLOWING_SYNC_KEY.$id, 1, 604800); + $this->markSynced($id, 'followers'); + $this->markSynced($id, 'following'); return; } - $hasFollowerPostProcessing = false; - $hasFollowingPostProcessing = false; + /* + * Keep these as separate queries. + * + * This lets the database use the indexes on profile_id and + * following_id independently instead of doing the previous OR scan. + */ + $followingCount = DB::table('followers') + ->where('profile_id', $id) + ->count(); + + $followersCount = DB::table('followers') + ->where('following_id', $id) + ->count(); + + /* + * Refresh the denormalized counters without hydrating the full model. + */ + Profile::query() + ->whereKey($id) + ->update([ + 'following_count' => $followingCount, + 'followers_count' => $followersCount, + ]); + + /* + * Only warm directions that are not already synced or currently + * being processed. + */ + if (! $followingReady) { + $this->warmDirection( + profileId: $id, + type: 'following', + count: $followingCount + ); + } - if (Follower::whereProfileId($id)->orWhere('following_id', $id)->count()) { - $following = []; - $followers = []; - foreach (Follower::where('following_id', $id)->orWhere('profile_id', $id)->lazyById(500) as $follow) { - if ($follow->following_id != $id && $follow->profile_id != $id) { - continue; - } - if ($follow->profile_id == $id) { - $following[] = $follow->following_id; - } else { - $followers[] = $follow->profile_id; - } - } + if (! $followersReady) { + $this->warmDirection( + profileId: $id, + type: 'followers', + count: $followersCount + ); + } - if (count($followers) > 100) { - // store follower ids and process in another job - Storage::put('follow-warm-cache/'.$id.'/followers.json', json_encode($followers)); - $hasFollowerPostProcessing = true; - } else { - foreach ($followers as $follower) { - FollowerService::add($follower, $id); - } - } + AccountService::del($id); + } - if (count($following) > 100) { - // store following ids and process in another job - Storage::put('follow-warm-cache/'.$id.'/following.json', json_encode($following)); - $hasFollowingPostProcessing = true; + private function warmDirection( + int $profileId, + string $type, + int $count + ): void { + if ($count === 0) { + $this->markSynced($profileId, $type); + + return; + } + + /* + * Tiny accounts can be handled immediately without the overhead + * of another queue job. + */ + if ($count <= self::INLINE_LIMIT) { + if ($type === 'followers') { + $ids = DB::table('followers') + ->where('following_id', $profileId) + ->orderBy('id') + ->limit(self::INLINE_LIMIT) + ->pluck('profile_id'); + + foreach ($ids as $followerId) { + FollowerService::add( + (int) $followerId, + $profileId + ); + } } else { - foreach ($following as $following) { - FollowerService::add($id, $following); + $ids = DB::table('followers') + ->where('profile_id', $profileId) + ->orderBy('id') + ->limit(self::INLINE_LIMIT) + ->pluck('following_id'); + + foreach ($ids as $followingId) { + FollowerService::add( + $profileId, + (int) $followingId + ); } } - } - Cache::put(FollowerService::FOLLOWERS_SYNC_KEY.$id, 1, 604800); - Cache::put(FollowerService::FOLLOWING_SYNC_KEY.$id, 1, 604800); + $this->markSynced($profileId, $type); - $profile = Profile::find($id); - if ($profile) { - $profile->following_count = DB::table('followers')->whereProfileId($id)->count(); - $profile->followers_count = DB::table('followers')->whereFollowingId($id)->count(); - $profile->save(); + return; } - AccountService::del($id); + /* + * Large graphs are paginated directly from the database. + * + * Cache::add() is atomic with Redis and prevents repeated + * FollowServiceWarmCache jobs from starting duplicate pipelines. + */ + $processingKey = + FollowServiceWarmCacheLargeIngestPipeline::processingKey( + $profileId, + $type + ); + + if (! Cache::add( + $processingKey, + 1, + self::PROCESSING_TTL + )) { + return; + } - if ($hasFollowingPostProcessing) { - FollowServiceWarmCacheLargeIngestPipeline::dispatch($id, 'following')->onQueue('follow'); + try { + FollowServiceWarmCacheLargeIngestPipeline::dispatch( + $profileId, + $type + )->onQueue('follow'); + } catch (Throwable $e) { + Cache::forget($processingKey); + + throw $e; } + } - if ($hasFollowerPostProcessing) { - FollowServiceWarmCacheLargeIngestPipeline::dispatch($id, 'followers')->onQueue('follow'); + private function isReady(int $profileId, string $type): bool + { + if (Cache::has($this->syncKey($profileId, $type))) { + return true; } + return Cache::has( + FollowServiceWarmCacheLargeIngestPipeline::processingKey( + $profileId, + $type + ) + ); + } + + private function markSynced(int $profileId, string $type): void + { + Cache::put( + $this->syncKey($profileId, $type), + 1, + self::SYNC_TTL + ); + } + + private function syncKey(int $profileId, string $type): string + { + return $type === 'followers' + ? FollowerService::FOLLOWERS_SYNC_KEY.$profileId + : FollowerService::FOLLOWING_SYNC_KEY.$profileId; } } diff --git a/app/Jobs/FollowPipeline/FollowServiceWarmCacheLargeIngestPipeline.php b/app/Jobs/FollowPipeline/FollowServiceWarmCacheLargeIngestPipeline.php index 0ab8ee161..46d0fddc9 100644 --- a/app/Jobs/FollowPipeline/FollowServiceWarmCacheLargeIngestPipeline.php +++ b/app/Jobs/FollowPipeline/FollowServiceWarmCacheLargeIngestPipeline.php @@ -8,79 +8,177 @@ use Illuminate\Contracts\Queue\ShouldQueue; use Illuminate\Foundation\Bus\Dispatchable; use Illuminate\Queue\InteractsWithQueue; use Illuminate\Queue\SerializesModels; -use Illuminate\Support\Facades\Storage; +use Illuminate\Support\Facades\Cache; +use Illuminate\Support\Facades\DB; +use InvalidArgumentException; class FollowServiceWarmCacheLargeIngestPipeline implements ShouldQueue { use Dispatchable, InteractsWithQueue, Queueable, SerializesModels; + private const CHUNK_SIZE = 1000; + + private const SYNC_TTL = 604800; // 7 days + + private const PROCESSING_TTL = 21600; // 6 hours + public $profileId; public $followType; + public $cursor; + public $tries = 5; - public $timeout = 5000; + public $timeout = 300; public $failOnTimeout = false; /** * Create a new job instance. - * - * @return void */ - public function __construct($profileId, $followType = 'following') - { + public function __construct( + $profileId, + $followType = 'following', + $cursor = 0 + ) { + if (! in_array($followType, ['followers', 'following'], true)) { + throw new InvalidArgumentException( + 'Invalid follow type: '.$followType + ); + } + $this->profileId = $profileId; $this->followType = $followType; + $this->cursor = $cursor; } /** * Execute the job. - * - * @return void */ - public function handle() + public function handle(): void { - $pid = $this->profileId; + $profileId = (int) $this->profileId; + $cursor = (int) $this->cursor; $type = $this->followType; + if ($profileId <= 0) { + return; + } + + /* + * Refresh the processing marker while the chain is active. + */ + Cache::put( + self::processingKey($profileId, $type), + 1, + self::PROCESSING_TTL + ); + if ($type === 'followers') { - $key = 'follow-warm-cache/'.$pid.'/followers.json'; - if (! Storage::exists($key)) { - return; - } - $file = Storage::get($key); - $json = json_decode($file, true); + $rows = DB::table('followers') + ->select([ + 'id', + 'profile_id', + ]) + ->where('following_id', $profileId) + ->where('id', '>', $cursor) + ->orderBy('id') + ->limit(self::CHUNK_SIZE) + ->get(); + } else { + $rows = DB::table('followers') + ->select([ + 'id', + 'following_id', + ]) + ->where('profile_id', $profileId) + ->where('id', '>', $cursor) + ->orderBy('id') + ->limit(self::CHUNK_SIZE) + ->get(); + } - foreach ($json as $id) { - FollowerService::add($id, $pid, false); - usleep(random_int(500, 3000)); - } - sleep(5); - Storage::delete($key); + /* + * Nothing left to process. + */ + if ($rows->isEmpty()) { + $this->complete($profileId, $type); + + return; } - if ($type === 'following') { - $key = 'follow-warm-cache/'.$pid.'/following.json'; - if (! Storage::exists($key)) { - return; - } - $file = Storage::get($key); - $json = json_decode($file, true); + foreach ($rows as $row) { + if ($type === 'followers') { + $followerId = (int) $row->profile_id; - foreach ($json as $id) { - FollowerService::add($pid, $id, false); - usleep(random_int(500, 3000)); + if ($followerId > 0) { + FollowerService::add( + $followerId, + $profileId, + false + ); + } + } else { + $followingId = (int) $row->following_id; + + if ($followingId > 0) { + FollowerService::add( + $profileId, + $followingId, + false + ); + } } - sleep(5); - Storage::delete($key); } - sleep(random_int(2, 5)); - $files = Storage::files('follow-warm-cache/'.$pid); - if (empty($files)) { - Storage::deleteDirectory('follow-warm-cache/'.$pid); + $lastRow = $rows->last(); + $nextCursor = (int) $lastRow->id; + + /* + * If we received fewer than CHUNK_SIZE rows, we know this was + * the final page and can finish without dispatching another job. + */ + if ($rows->count() < self::CHUNK_SIZE) { + $this->complete($profileId, $type); + + return; } + + /* + * One bounded chunk per queue job. + * + * Memory usage remains effectively constant regardless of whether + * the account has 1,000 or 10,000,000 followers. + */ + self::dispatch( + $profileId, + $type, + $nextCursor + )->onQueue('follow'); + } + + public static function processingKey( + int $profileId, + string $type + ): string { + return 'pf:follow-warm-cache:processing:'.$profileId.':'.$type; + } + + private function complete(int $profileId, string $type): void + { + $syncKey = $type === 'followers' + ? FollowerService::FOLLOWERS_SYNC_KEY.$profileId + : FollowerService::FOLLOWING_SYNC_KEY.$profileId; + + Cache::put( + $syncKey, + 1, + self::SYNC_TTL + ); + + Cache::forget( + self::processingKey($profileId, $type) + ); } }