diff --git a/app/Exceptions/InvalidDeliveryDestinationException.php b/app/Exceptions/InvalidDeliveryDestinationException.php new file mode 100644 index 000000000..74a6b7f29 --- /dev/null +++ b/app/Exceptions/InvalidDeliveryDestinationException.php @@ -0,0 +1,14 @@ +sender); + $domain = DeliveryHostService::domain($this->to); + $url = self::validateDestination($this->to); if (! $url) { - throw new InvalidArgumentException( + if ($domain) { + DeliveryHostService::recordFailure($domain); + } + + throw new InvalidDeliveryDestinationException( 'Invalid ActivityPub destination URL.' ); } @@ -91,6 +101,15 @@ class ActivityPubDeliveryService return; } + if ($domain && DeliveryHostService::isUnavailable($domain)) { + Log::info('Skipped ActivityPub delivery to unavailable host', [ + 'profile_id' => $this->sender->id, + 'url' => $url, + ]); + + return; + } + try { $payload = self::serializePayload($this->payload); @@ -106,6 +125,14 @@ class ActivityPubDeliveryService $headers ); + if ($domain) { + if ($response->serverError()) { + DeliveryHostService::recordFailure($domain); + } else { + DeliveryHostService::recordSuccess($domain); + } + } + if ($response->failed()) { self::logFailedResponse( $url, @@ -114,6 +141,10 @@ class ActivityPubDeliveryService ); } } catch (Throwable $e) { + if ($domain && $e instanceof ConnectionException) { + DeliveryHostService::recordFailure($domain); + } + Log::warning('ActivityPub delivery failed', [ 'profile_id' => $this->sender->id, 'url' => $url, @@ -132,10 +163,16 @@ class ActivityPubDeliveryService * is both signed and transmitted, ensuring the Digest header matches the * bytes received by the remote ActivityPub server. * + * Hosts currently marked unavailable by DeliveryHostService are skipped + * silently (counted in the result, no $onError call). Connection + * failures, 5xx responses and inbox URLs that fail validation count + * against the host; any other response clears its failure count. + * * @param Profile $profile Local sender used for HTTP signatures * @param array $audience Inbox URLs * @param array $activity ActivityPub activity * @param \Closure|null $onError fn(Throwable|Response $reason, int $index): void + * @return array{total: int, skipped: int, duplicate: int, invalid: int, sent: int, delivered: int, rejected: int, failed: int} * * @throws JsonException */ @@ -144,9 +181,20 @@ class ActivityPubDeliveryService array $audience, array $activity, ?\Closure $onError = null - ): void { + ): array { + $result = [ + 'total' => count($audience), + 'skipped' => 0, // host currently marked unavailable + 'duplicate' => 0, // same inbox URL earlier in this audience + 'invalid' => 0, // failed validation (dead DNS, banned, malformed) + 'sent' => 0, // requests actually made + 'delivered' => 0, // 2xx / 3xx + 'rejected' => 0, // 4xx / 5xx + 'failed' => 0, // connection failure or signing error + ]; + if (empty($audience)) { - return; + return $result; } self::validateSender($profile); @@ -157,7 +205,7 @@ class ActivityPubDeliveryService 'destinations' => count($audience), ]); - return; + return $result; } /* @@ -178,18 +226,36 @@ class ActivityPubDeliveryService $seen = []; + /* + * Host health, collected during the batch and applied once at the + * end so no database writes happen inside the HTTP phase. + */ + $hostFailures = []; + + $hostSuccesses = []; + foreach ($audience as $index => $destination) { + $domain = null; + try { - if (! is_string($destination) || $destination === '') { - throw new InvalidArgumentException( + if (! is_string($destination) || trim($destination) === '') { + throw new InvalidDeliveryDestinationException( 'ActivityPub inbox URL must be a non-empty string.' ); } + $domain = DeliveryHostService::domain($destination); + + if ($domain && DeliveryHostService::isUnavailable($domain)) { + $result['skipped']++; + + continue; + } + $url = self::validateDestination($destination); if (! $url) { - throw new InvalidArgumentException( + throw new InvalidDeliveryDestinationException( 'Invalid ActivityPub destination URL.' ); } @@ -200,6 +266,8 @@ class ActivityPubDeliveryService * in duplicate URLs. */ if (isset($seen[$url])) { + $result['duplicate']++; + continue; } @@ -214,9 +282,40 @@ class ActivityPubDeliveryService $deliveries[] = [ 'index' => $index, 'url' => $url, + 'domain' => DeliveryHostService::domain($url) ?? $domain, 'headers' => $headers, ]; + } catch (InvalidDeliveryDestinationException $e) { + /* + * Expected churn: dead hosts, banned instances, stale rows. + * Counted against the host and reported to the caller, but + * not worth a warning per inbox per activity. + */ + $result['invalid']++; + + if ($domain) { + $hostFailures[$domain] = true; + } + + Log::debug('Skipped ActivityPub delivery to invalid inbox', [ + 'profile_id' => $profile->id, + 'index' => $index, + 'url' => is_string($destination) + ? $destination + : null, + 'error' => $e->getMessage(), + ]); + + if ($onError) { + $onError($e, $index); + } } catch (Throwable $e) { + /* + * Anything else here is a signing or serialization problem + * on our side and deserves attention. + */ + $result['failed']++; + Log::warning('Unable to prepare ActivityPub delivery', [ 'profile_id' => $profile->id, 'index' => $index, @@ -233,100 +332,147 @@ class ActivityPubDeliveryService } } - if (empty($deliveries)) { - return; - } + $result['sent'] = count($deliveries); - $timeout = self::deliveryTimeout(); - $connectTimeout = self::connectTimeout($timeout); + if (! empty($deliveries)) { + $timeout = self::deliveryTimeout(); + $connectTimeout = self::connectTimeout($timeout); - /* - * Each request is named using the original audience index. - * - * Laravel's Pool::as() ensures the response can be mapped directly - * back to that destination even when some audience entries were - * rejected during validation/signing. - */ - $responses = Http::pool( - function (HttpPool $pool) use ( - $deliveries, - $payload, - $timeout, - $connectTimeout - ) { - foreach ($deliveries as $delivery) { - $pool - ->as((string) $delivery['index']) - ->replaceHeaders($delivery['headers']) - ->timeout($timeout) - ->connectTimeout($connectTimeout) - ->withoutRedirecting() - ->send('POST', $delivery['url'], [ - 'body' => $payload, - ]); + /* + * Each request is named using the original audience index. + * + * Laravel's Pool::as() ensures the response can be mapped directly + * back to that destination even when some audience entries were + * rejected during validation/signing. + */ + $responses = Http::pool( + function (HttpPool $pool) use ( + $deliveries, + $payload, + $timeout, + $connectTimeout + ) { + foreach ($deliveries as $delivery) { + $pool + ->as((string) $delivery['index']) + ->replaceHeaders($delivery['headers']) + ->timeout($timeout) + ->connectTimeout($connectTimeout) + ->withoutRedirecting() + ->send('POST', $delivery['url'], [ + 'body' => $payload, + ]); + } } + ); + + $deliveriesByIndex = []; + + foreach ($deliveries as $delivery) { + $deliveriesByIndex[(string) $delivery['index']] = $delivery; } - ); - $deliveriesByIndex = []; + foreach ($responses as $index => $response) { + $delivery = $deliveriesByIndex[(string) $index] ?? null; - foreach ($deliveries as $delivery) { - $deliveriesByIndex[(string) $delivery['index']] = $delivery; - } + $url = $delivery['url'] ?? null; - foreach ($responses as $index => $response) { - $delivery = $deliveriesByIndex[(string) $index] ?? null; + $domain = $delivery['domain'] ?? null; - $url = $delivery['url'] ?? null; + if ($response instanceof Throwable) { + $result['failed']++; - if ($response instanceof Throwable) { - Log::warning('ActivityPub pooled delivery connection failure', [ - 'profile_id' => $profile->id, - 'index' => $index, - 'url' => $url, - 'exception' => $response::class, - 'error' => $response->getMessage(), - ]); + if ($domain) { + $hostFailures[$domain] = true; + } - if ($onError) { - $onError($response, (int) $index); + Log::info('ActivityPub pooled delivery connection failure', [ + 'profile_id' => $profile->id, + 'index' => $index, + 'url' => $url, + 'exception' => $response::class, + 'error' => $response->getMessage(), + ]); + + if ($onError) { + $onError($response, (int) $index); + } + + continue; } - continue; - } + if (! $response instanceof Response) { + $result['failed']++; - if (! $response instanceof Response) { - $exception = new RuntimeException( - 'Unexpected ActivityPub HTTP pool response type.' - ); + $exception = new RuntimeException( + 'Unexpected ActivityPub HTTP pool response type.' + ); - Log::warning('Unexpected ActivityPub pooled delivery response', [ - 'profile_id' => $profile->id, - 'index' => $index, - 'url' => $url, - 'response_type' => get_debug_type($response), - ]); + Log::warning('Unexpected ActivityPub pooled delivery response', [ + 'profile_id' => $profile->id, + 'index' => $index, + 'url' => $url, + 'response_type' => get_debug_type($response), + ]); - if ($onError) { - $onError($exception, (int) $index); + if ($onError) { + $onError($exception, (int) $index); + } + + continue; } - continue; - } + if ($response->failed()) { + $result['rejected']++; + + if ($domain) { + /* + * 5xx means the host is broken or dead behind a + * proxy. 4xx means it answered, so it is reachable + * even if it disliked the request. + */ + if ($response->serverError()) { + $hostFailures[$domain] = true; + } else { + $hostSuccesses[$domain] = true; + } + } + + self::logFailedResponse( + $url, + $profile, + $response, + (int) $index + ); - if ($response->failed()) { - self::logFailedResponse( - $url, - $profile, - $response, - (int) $index - ); + if ($onError) { + $onError($response, (int) $index); + } - if ($onError) { - $onError($response, (int) $index); + continue; + } + + $result['delivered']++; + + if ($domain) { + $hostSuccesses[$domain] = true; } } } + + /* + * A host that answered at all during this batch is reachable, even + * if another inbox on it failed. + */ + foreach (array_keys($hostSuccesses) as $domain) { + unset($hostFailures[$domain]); + } + + DeliveryHostService::recordFailures(array_keys($hostFailures)); + + DeliveryHostService::recordSuccesses(array_keys($hostSuccesses)); + + return $result; } /** @@ -605,7 +751,13 @@ class ActivityPubDeliveryService $context['index'] = $index; } - Log::warning( + /* + * 4xx usually means a signature or compatibility problem worth + * seeing. 5xx is almost always a broken or dead host and is handled + * by DeliveryHostService, so keep it out of the warning stream. + */ + Log::log( + $response->serverError() ? 'info' : 'warning', 'ActivityPub delivery rejected by remote server', $context ); diff --git a/app/Services/DeliveryHostService.php b/app/Services/DeliveryHostService.php new file mode 100644 index 000000000..254490f9f --- /dev/null +++ b/app/Services/DeliveryHostService.php @@ -0,0 +1,364 @@ +|null */ + private static ?array $memo = null; + + private static int $memoAt = 0; + + /** + * Lowercase host of an inbox URL, or null if it cannot be parsed. + */ + public static function domain(?string $url): ?string + { + if (! is_string($url) || trim($url) === '') { + return null; + } + + $host = parse_url(trim($url), PHP_URL_HOST); + + if (! is_string($host) || $host === '') { + return null; + } + + return strtolower($host); + } + + public static function isUnavailable(string $domain): bool + { + $next = self::flagged()[strtolower($domain)] ?? null; + + return $next !== null && $next > time(); + } + + /** + * Domains currently being skipped. + * + * @return array + */ + public static function unavailable(): array + { + $now = time(); + + return array_keys( + array_filter( + self::flagged(), + fn (?int $next) => $next !== null && $next > $now + ) + ); + } + + /** + * Drop inbox URLs whose host is unavailable, plus any null or empty + * entries. Keys are preserved so callers can map back to the original + * audience index. + * + * @param array $inboxes + * @return array + */ + public static function filter(array $inboxes): array + { + $kept = []; + + foreach ($inboxes as $index => $inbox) { + if (! is_string($inbox) || trim($inbox) === '') { + continue; + } + + $domain = self::domain($inbox); + + if ($domain && self::isUnavailable($domain)) { + continue; + } + + $kept[$index] = $inbox; + } + + return $kept; + } + + public static function recordFailure(string $domain): void + { + self::recordFailures([$domain]); + } + + /** + * @param array $domains + */ + public static function recordFailures(array $domains): void + { + $domains = self::normalize($domains); + + if (empty($domains)) { + return; + } + + foreach ($domains as $domain) { + self::applyFailure($domain); + } + + self::flush(); + } + + public static function recordSuccess(string $domain): void + { + self::recordSuccesses([$domain]); + } + + /** + * Only hosts with an existing failure count are touched, so this costs + * nothing for the healthy majority of a fanout. + * + * @param array $domains + */ + public static function recordSuccesses(array $domains): void + { + $flagged = self::flagged(); + + $domains = array_filter( + self::normalize($domains), + fn (string $domain) => array_key_exists($domain, $flagged) + ); + + if (empty($domains)) { + return; + } + + self::clear(array_values($domains)); + } + + /** + * Clear a host's failure state manually (admin / tinker). + */ + public static function reset(string $domain): void + { + $domains = self::normalize([$domain]); + + if (empty($domains)) { + return; + } + + self::clear(array_values($domains)); + } + + /** + * @param array $domains + */ + private static function clear(array $domains): void + { + try { + Instance::query() + ->whereIn('domain', $domains) + ->update([ + 'delivery_failures' => 0, + 'delivery_timeout' => false, + 'delivery_next_after' => null, + ]); + } catch (Throwable $e) { + Log::warning('Unable to clear delivery failures', [ + 'domains' => $domains, + 'exception' => $e::class, + 'error' => $e->getMessage(), + ]); + } + + self::flush(); + } + + private static function applyFailure(string $domain): void + { + try { + $instance = Instance::whereDomain($domain)->first(); + + if (! $instance) { + /* + * Explicit assignment rather than firstOrCreate() so this + * does not depend on Instance::$fillable. + */ + $instance = new Instance; + $instance->domain = $domain; + $instance->save(); + } + + $instance->increment('delivery_failures'); + + $failures = (int) $instance->delivery_failures; + $threshold = self::threshold(); + + if ($failures < $threshold) { + return; + } + + $instance->forceFill([ + 'delivery_timeout' => true, + 'delivery_next_after' => Carbon::now()->addSeconds( + self::backoff($failures - $threshold) + ), + ])->save(); + } catch (Throwable $e) { + Log::warning('Unable to record delivery failure', [ + 'domain' => $domain, + 'exception' => $e::class, + 'error' => $e->getMessage(), + ]); + } + } + + /** + * Domains with a non-zero failure count, mapped to the unix timestamp + * after which delivery may be retried, or null while still under the + * threshold. + * + * @return array + */ + private static function flagged(): array + { + if ( + self::$memo !== null + && (time() - self::$memoAt) < self::MEMO_TTL + ) { + return self::$memo; + } + + self::$memoAt = time(); + + try { + return self::$memo = Cache::remember( + self::CACHE_KEY, + self::CACHE_TTL, + function () { + return Instance::query() + ->where('delivery_failures', '>', 0) + ->get([ + 'domain', + 'delivery_timeout', + 'delivery_next_after', + ]) + ->mapWithKeys(function ($instance) { + $next = null; + + if ( + $instance->delivery_timeout + && $instance->delivery_next_after + ) { + $next = Carbon::parse( + $instance->delivery_next_after + )->getTimestamp(); + } + + return [strtolower($instance->domain) => $next]; + }) + ->all(); + } + ); + } catch (Throwable $e) { + /* + * Never let host tracking break delivery. Memoize the empty + * result so a broken setup (e.g. migration not run) is logged + * once per window rather than once per inbox. + */ + Log::warning('Unable to load delivery host state', [ + 'exception' => $e::class, + 'error' => $e->getMessage(), + ]); + + return self::$memo = []; + } + } + + private static function flush(): void + { + self::$memo = null; + self::$memoAt = 0; + + Cache::forget(self::CACHE_KEY); + } + + /** + * @param array $domains + * @return array Deduplicated, keyed by domain + */ + private static function normalize(array $domains): array + { + $out = []; + + foreach ($domains as $domain) { + if (! is_string($domain)) { + continue; + } + + $domain = strtolower(trim($domain)); + + if ($domain === '') { + continue; + } + + $out[$domain] = $domain; + } + + return $out; + } + + private static function threshold(): int + { + return max( + 1, + (int) config('federation.activitypub.delivery.failure_threshold', 5) + ); + } + + /** + * Seconds to wait before retrying, doubling with each failure past the + * threshold: 1h, 2h, 4h ... capped at max_backoff. + */ + private static function backoff(int $excess): int + { + $max = max( + self::BASE_BACKOFF, + (int) config('federation.activitypub.delivery.max_backoff', 604800) + ); + + $exponent = min(max($excess, 0), self::MAX_BACKOFF_EXPONENT); + + return (int) min(self::BASE_BACKOFF * (2 ** $exponent), $max); + } +} diff --git a/config/federation.php b/config/federation.php index c424d3a4e..a409650c5 100644 --- a/config/federation.php +++ b/config/federation.php @@ -25,6 +25,8 @@ return [ 'enabled' => env('AP_LOGGER_ENABLED', false), 'driver' => 'log', ], + 'failure_threshold' => env('AP_DELIVERY_FAILURE_THRESHOLD', 5), + 'max_backoff' => env('AP_DELIVERY_MAX_BACKOFF', 604800), ], 'ingest' => [ diff --git a/database/migrations/2026_09_12_104611_add_instances_delivery_failures_migration.php b/database/migrations/2026_09_12_104611_add_instances_delivery_failures_migration.php new file mode 100644 index 000000000..c38e378ed --- /dev/null +++ b/database/migrations/2026_09_12_104611_add_instances_delivery_failures_migration.php @@ -0,0 +1,34 @@ +boolean('delivery_timeout')->default(false)->index(); + } + + if (! Schema::hasColumn('instances', 'delivery_next_after')) { + $table->timestamp('delivery_next_after')->nullable(); + } + + if (! Schema::hasColumn('instances', 'delivery_failures')) { + $table->unsignedSmallInteger('delivery_failures')->default(0)->index(); + } + }); + } + + public function down(): void + { + Schema::table('instances', function (Blueprint $table) { + if (Schema::hasColumn('instances', 'delivery_failures')) { + $table->dropColumn('delivery_failures'); + } + }); + } +};