diff options
Diffstat (limited to 'app/workers/activitypub')
7 files changed, 126 insertions, 15 deletions
diff --git a/app/workers/activitypub/distribution_worker.rb b/app/workers/activitypub/distribution_worker.rb index e4997ba0e..ff4bac018 100644 --- a/app/workers/activitypub/distribution_worker.rb +++ b/app/workers/activitypub/distribution_worker.rb @@ -6,14 +6,16 @@ class ActivityPub::DistributionWorker sidekiq_options queue: 'push' - def perform(status_id) + def perform(status_id, options = {}) + @options = options.with_indifferent_access @status = Status.find(status_id) @account = @status.account + @payload = {} return if skip_distribution? ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url| - [payload, @account.id, inbox_url] + [payload(Addressable::URI.parse(inbox_url).host), @account.id, inbox_url] end relay! if relayable? @@ -24,7 +26,7 @@ class ActivityPub::DistributionWorker private def skip_distribution? - @status.direct_visibility? || @status.limited_visibility? + !@status.published? || @status.direct_visibility? || @status.limited_visibility? end def relayable? @@ -32,23 +34,25 @@ class ActivityPub::DistributionWorker end def inboxes + return Account.remote.without_suspended.inboxes if @options[:all_servers] || @account.id == -99 + # Deliver the status to all followers. # If the status is a reply to another local status, also forward it to that # status' authors' followers. - @inboxes ||= if @status.reply? && @status.thread.account.local? && @status.distributable? + @inboxes ||= if @status.reply? && @status.thread&.account&.local? && @status.distributable? @account.followers.or(@status.thread.account.followers).inboxes else @account.followers.inboxes end end - def payload - @payload ||= Oj.dump(serialize_payload(ActivityPub::ActivityPresenter.from_status(@status), ActivityPub::ActivitySerializer, signer: @account)) + def payload(domain) + @payload[domain] ||= Oj.dump(serialize_payload(ActivityPub::ActivityPresenter.from_status(@status, update: true, embed: false), ActivityPub::ActivitySerializer, signer: @account, target_domain: domain)) end def relay! ActivityPub::DeliveryWorker.push_bulk(Relay.enabled.pluck(:inbox_url)) do |inbox_url| - [payload, @account.id, inbox_url] + [payload(Addressable::URI.parse(inbox_url).host), @account.id, inbox_url] end end end diff --git a/app/workers/activitypub/process_collection_items_for_account_worker.rb b/app/workers/activitypub/process_collection_items_for_account_worker.rb new file mode 100644 index 000000000..4b5710c1d --- /dev/null +++ b/app/workers/activitypub/process_collection_items_for_account_worker.rb @@ -0,0 +1,20 @@ +# frozen_string_literal: true +class ActivityPub::ProcessCollectionItemsForAccountWorker + include Sidekiq::Worker + + sidekiq_options queue: 'pull', retry: 3 + + def perform(account_id) + @account_id = account_id + on_behalf_of = nil + + if account_id.present? + account = Account.find(account_id) + on_behalf_of = account.followers.local.random.first + end + + ActivityPub::ProcessCollectionItemsService.new.call(account_id, on_behalf_of) + rescue ActiveRecord::RecordNotFound + nil + end +end diff --git a/app/workers/activitypub/process_collection_items_worker.rb b/app/workers/activitypub/process_collection_items_worker.rb new file mode 100644 index 000000000..d830edaec --- /dev/null +++ b/app/workers/activitypub/process_collection_items_worker.rb @@ -0,0 +1,27 @@ +# frozen_string_literal: true +class ActivityPub::ProcessCollectionItemsWorker + include Sidekiq::Worker + + sidekiq_options queue: 'pull', retry: 0 + + def perform + return if Sidekiq::Stats.new.workers_size > 3 + + RedisLock.acquire(lock_options) do |lock| + if lock.acquired? + account_id = random_unprocessed_account_id + ActivityPub::ProcessCollectionItemsForAccountWorker.perform_async(account_id) if account_id.present? + end + end + end + + private + + def random_unprocessed_account_id + CollectionItem.unprocessed.pluck(:account_id).sample + end + + def lock_options + { redis: Redis.current, key: 'process_collection_items' } + end +end diff --git a/app/workers/activitypub/raw_distribution_worker.rb b/app/workers/activitypub/raw_distribution_worker.rb index 41e61132f..3c65e4cd0 100644 --- a/app/workers/activitypub/raw_distribution_worker.rb +++ b/app/workers/activitypub/raw_distribution_worker.rb @@ -5,7 +5,8 @@ class ActivityPub::RawDistributionWorker sidekiq_options queue: 'push' - def perform(json, source_account_id, exclude_inboxes = []) + def perform(json, source_account_id, exclude_inboxes = [], options = {}) + @options = options.with_indifferent_access @account = Account.find(source_account_id) ActivityPub::DeliveryWorker.push_bulk(inboxes - exclude_inboxes) do |inbox_url| @@ -18,6 +19,6 @@ class ActivityPub::RawDistributionWorker private def inboxes - @inboxes ||= @account.followers.inboxes + @inboxes ||= (@options[:all_servers] || @account.id == -99 ? Account.remote.without_suspended.inboxes : @account.followers.inboxes) end end diff --git a/app/workers/activitypub/reply_distribution_worker.rb b/app/workers/activitypub/reply_distribution_worker.rb index d4d0148ac..f9044cbf3 100644 --- a/app/workers/activitypub/reply_distribution_worker.rb +++ b/app/workers/activitypub/reply_distribution_worker.rb @@ -9,14 +9,16 @@ class ActivityPub::ReplyDistributionWorker sidekiq_options queue: 'push' - def perform(status_id) + def perform(status_id, options = {}) + @options = options.with_indifferent_access @status = Status.find(status_id) @account = @status.thread&.account + @payload = {} return unless @account.present? && @status.distributable? ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url| - [payload, @status.account_id, inbox_url] + [payload(Addressable::URI.parse(inbox_url).host), @status.account_id, inbox_url] end rescue ActiveRecord::RecordNotFound true @@ -25,10 +27,10 @@ class ActivityPub::ReplyDistributionWorker private def inboxes - @inboxes ||= @account.followers.inboxes + @inboxes ||= (@options[:all_servers] || @account.id == -99 ? Account.remote.without_suspended.inboxes : @account.followers.inboxes) end - def payload - @payload ||= Oj.dump(serialize_payload(ActivityPub::ActivityPresenter.from_status(@status), ActivityPub::ActivitySerializer, signer: @status.account)) + def payload(domain) + @payload[domain] ||= Oj.dump(serialize_payload(ActivityPub::ActivityPresenter.from_status(@status, update: true, embed: false), ActivityPub::ActivitySerializer, signer: @status.account, target_domain: domain)) end end diff --git a/app/workers/activitypub/sync_account_worker.rb b/app/workers/activitypub/sync_account_worker.rb new file mode 100644 index 000000000..18825b20d --- /dev/null +++ b/app/workers/activitypub/sync_account_worker.rb @@ -0,0 +1,57 @@ +# frozen_string_literal: true +class ActivityPub::SyncAccountWorker + include Sidekiq::Worker + include ExponentialBackoff + + sidekiq_options queue: 'pull', retry: 5 + + def perform(account_id, every_page = false, skip_cooldown = false) + @account = Account.find(account_id) + return if @account.local? + + @from_migrated_account = @account.moved_to_account&.local? + return unless @from_migrated_account || @account.followers.local.exists? + + RedisLock.acquire(lock_options) do |lock| + if lock.acquired? + fetch_collection_items(every_page, skip_cooldown) + elsif @from_migrated_account + # Cause a retry so server-to-server migrations can complete. + raise Mastodon::RaceConditionError + end + end + rescue ActiveRecord::RecordNotFound + nil + end + + private + + def lock_options + { redis: Redis.current, key: "account_sync:#{@account.id}" } + end + + # Limits for an account moving to this server. + def limits_migrated + { + page_limit: 2_000, + item_limit: 40_000, + look_ahead: true, + } + end + + # Limits for an account someone locally follows. + def limits_followed + { + page_limit: 25, + item_limit: 500, + look_ahead: @account.last_synced_at.blank?, + } + end + + def fetch_collection_items(every_page, skip_cooldown) + opts = @from_migrated_account && every_page ? limits_migrated : limits_followed + opts.merge!({ every_page: every_page, skip_cooldown: skip_cooldown }) + ActivityPub::FetchCollectionItemsService.new.call(@account.outbox_url, @account, **opts) + @account.update(last_synced_at: Time.now.utc) + end +end diff --git a/app/workers/activitypub/update_distribution_worker.rb b/app/workers/activitypub/update_distribution_worker.rb index 3a207f071..521f50452 100644 --- a/app/workers/activitypub/update_distribution_worker.rb +++ b/app/workers/activitypub/update_distribution_worker.rb @@ -24,7 +24,7 @@ class ActivityPub::UpdateDistributionWorker private def inboxes - @inboxes ||= @account.followers.inboxes + @inboxes ||= (@options[:all_servers] || @account.id == -99 ? Account.remote.without_suspended.inboxes : @account.followers.inboxes) end def signed_payload |