diff options
author | ThibG <thib@sitedethib.com> | 2019-07-21 00:53:28 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2019-07-21 00:53:28 +0200 |
commit | f1597e1ab90b1fb291f16977877c6ca79bf89676 (patch) | |
tree | 7cb8facbd66c2199e173d2fc897863108b1fb398 /app/workers | |
parent | 19c3a941e8ac53a6e1ab6e9c0aaadbc53dd0050f (diff) | |
parent | 055450bc02732f654317096f56b03b77f5b84270 (diff) |
Merge pull request #1158 from ThibG/glitch-soc/merge-upstream
Merge upstream changes
Diffstat (limited to 'app/workers')
-rw-r--r-- | app/workers/activitypub/delivery_worker.rb | 30 | ||||
-rw-r--r-- | app/workers/after_remote_follow_request_worker.rb | 24 | ||||
-rw-r--r-- | app/workers/after_remote_follow_worker.rb | 24 | ||||
-rw-r--r-- | app/workers/maintenance/uncache_preview_worker.rb | 18 | ||||
-rw-r--r-- | app/workers/notification_worker.rb | 4 | ||||
-rw-r--r-- | app/workers/processing_worker.rb | 4 | ||||
-rw-r--r-- | app/workers/pubsubhubbub/confirmation_worker.rb | 75 | ||||
-rw-r--r-- | app/workers/pubsubhubbub/delivery_worker.rb | 74 | ||||
-rw-r--r-- | app/workers/pubsubhubbub/distribution_worker.rb | 25 | ||||
-rw-r--r-- | app/workers/pubsubhubbub/raw_distribution_worker.rb | 15 | ||||
-rw-r--r-- | app/workers/pubsubhubbub/subscribe_worker.rb | 27 | ||||
-rw-r--r-- | app/workers/pubsubhubbub/unsubscribe_worker.rb | 8 | ||||
-rw-r--r-- | app/workers/remote_profile_update_worker.rb | 6 | ||||
-rw-r--r-- | app/workers/salmon_worker.rb | 6 | ||||
-rw-r--r-- | app/workers/scheduler/preview_cards_cleanup_scheduler.rb | 22 | ||||
-rw-r--r-- | app/workers/scheduler/subscriptions_scheduler.rb | 10 |
16 files changed, 67 insertions, 305 deletions
diff --git a/app/workers/activitypub/delivery_worker.rb b/app/workers/activitypub/delivery_worker.rb index 818fd8f5d..5457d9d4b 100644 --- a/app/workers/activitypub/delivery_worker.rb +++ b/app/workers/activitypub/delivery_worker.rb @@ -2,6 +2,7 @@ class ActivityPub::DeliveryWorker include Sidekiq::Worker + include JsonLdHelper STOPLIGHT_FAILURE_THRESHOLD = 10 STOPLIGHT_COOLDOWN = 60 @@ -18,21 +19,24 @@ class ActivityPub::DeliveryWorker @source_account = Account.find(source_account_id) @inbox_url = inbox_url @host = Addressable::URI.parse(inbox_url).normalized_site + @performed = false perform_request - - failure_tracker.track_success! - rescue => e - failure_tracker.track_failure! - raise e.class, "Delivery failed for #{inbox_url}: #{e.message}", e.backtrace[0] + ensure + if @performed + failure_tracker.track_success! + else + failure_tracker.track_failure! + end end private def build_request(http_client) - request = Request.new(:post, @inbox_url, body: @json, http_client: http_client) - request.on_behalf_of(@source_account, :uri, sign_with: @options[:sign_with]) - request.add_headers(HEADERS) + Request.new(:post, @inbox_url, body: @json, http_client: http_client).tap do |request| + request.on_behalf_of(@source_account, :uri, sign_with: @options[:sign_with]) + request.add_headers(HEADERS) + end end def perform_request @@ -40,6 +44,8 @@ class ActivityPub::DeliveryWorker request_pool.with(@host) do |http_client| build_request(http_client).perform do |response| raise Mastodon::UnexpectedResponseError, response unless response_successful?(response) || response_error_unsalvageable?(response) + + @performed = true end end end @@ -49,14 +55,6 @@ class ActivityPub::DeliveryWorker .run end - def response_successful?(response) - (200...300).cover?(response.code) - end - - def response_error_unsalvageable?(response) - response.code == 501 || ((400...500).cover?(response.code) && ![401, 408, 429].include?(response.code)) - end - def failure_tracker @failure_tracker ||= DeliveryFailureTracker.new(@inbox_url) end diff --git a/app/workers/after_remote_follow_request_worker.rb b/app/workers/after_remote_follow_request_worker.rb index 84eb6ade2..ce9c65834 100644 --- a/app/workers/after_remote_follow_request_worker.rb +++ b/app/workers/after_remote_follow_request_worker.rb @@ -5,27 +5,5 @@ class AfterRemoteFollowRequestWorker sidekiq_options queue: 'pull', retry: 5 - attr_reader :follow_request - - def perform(follow_request_id) - @follow_request = FollowRequest.find(follow_request_id) - process_follow_service if processing_required? - rescue ActiveRecord::RecordNotFound - true - end - - private - - def process_follow_service - follow_request.destroy - FollowService.new.call(follow_request.account, updated_account.acct) - end - - def processing_required? - !updated_account.nil? && !updated_account.locked? - end - - def updated_account - @_updated_account ||= FetchRemoteAccountService.new.call(follow_request.target_account.remote_url) - end + def perform(follow_request_id); end end diff --git a/app/workers/after_remote_follow_worker.rb b/app/workers/after_remote_follow_worker.rb index edab83f85..d9719f2bf 100644 --- a/app/workers/after_remote_follow_worker.rb +++ b/app/workers/after_remote_follow_worker.rb @@ -5,27 +5,5 @@ class AfterRemoteFollowWorker sidekiq_options queue: 'pull', retry: 5 - attr_reader :follow - - def perform(follow_id) - @follow = Follow.find(follow_id) - process_follow_service if processing_required? - rescue ActiveRecord::RecordNotFound - true - end - - private - - def process_follow_service - follow.destroy - FollowService.new.call(follow.account, updated_account.acct) - end - - def updated_account - @_updated_account ||= FetchRemoteAccountService.new.call(follow.target_account.remote_url) - end - - def processing_required? - !updated_account.nil? && updated_account.locked? - end + def perform(follow_id); end end diff --git a/app/workers/maintenance/uncache_preview_worker.rb b/app/workers/maintenance/uncache_preview_worker.rb new file mode 100644 index 000000000..810ffd8cc --- /dev/null +++ b/app/workers/maintenance/uncache_preview_worker.rb @@ -0,0 +1,18 @@ +# frozen_string_literal: true + +class Maintenance::UncachePreviewWorker + include Sidekiq::Worker + + sidekiq_options queue: 'pull' + + def perform(preview_card_id) + preview_card = PreviewCard.find(preview_card_id) + + return if preview_card.image.blank? + + preview_card.image.destroy + preview_card.save + rescue ActiveRecord::RecordNotFound + true + end +end diff --git a/app/workers/notification_worker.rb b/app/workers/notification_worker.rb index da1d6ab45..1c0f001cf 100644 --- a/app/workers/notification_worker.rb +++ b/app/workers/notification_worker.rb @@ -5,7 +5,5 @@ class NotificationWorker sidekiq_options queue: 'push', retry: 5 - def perform(xml, source_account_id, target_account_id) - SendInteractionService.new.call(xml, Account.find(source_account_id), Account.find(target_account_id)) - end + def perform(xml, source_account_id, target_account_id); end end diff --git a/app/workers/processing_worker.rb b/app/workers/processing_worker.rb index 978c3aba2..cf3bd8397 100644 --- a/app/workers/processing_worker.rb +++ b/app/workers/processing_worker.rb @@ -5,7 +5,5 @@ class ProcessingWorker sidekiq_options backtrace: true - def perform(account_id, body) - ProcessFeedService.new.call(body, Account.find(account_id), override_timestamps: true) - end + def perform(account_id, body); end end diff --git a/app/workers/pubsubhubbub/confirmation_worker.rb b/app/workers/pubsubhubbub/confirmation_worker.rb index c0e7b677e..783a8c95f 100644 --- a/app/workers/pubsubhubbub/confirmation_worker.rb +++ b/app/workers/pubsubhubbub/confirmation_worker.rb @@ -2,81 +2,8 @@ class Pubsubhubbub::ConfirmationWorker include Sidekiq::Worker - include RoutingHelper sidekiq_options queue: 'push', retry: false - attr_reader :subscription, :mode, :secret, :lease_seconds - - def perform(subscription_id, mode, secret = nil, lease_seconds = nil) - @subscription = Subscription.find(subscription_id) - @mode = mode - @secret = secret - @lease_seconds = lease_seconds - process_confirmation - end - - private - - def process_confirmation - prepare_subscription - - callback_get_with_params - logger.debug "Confirming PuSH subscription for #{subscription.callback_url} with challenge #{challenge}: #{@callback_response_body}" - - update_subscription - end - - def update_subscription - if successful_subscribe? - subscription.save! - elsif successful_unsubscribe? - subscription.destroy! - end - end - - def successful_subscribe? - subscribing? && response_matches_challenge? - end - - def successful_unsubscribe? - (unsubscribing? && response_matches_challenge?) || !subscription.confirmed? - end - - def response_matches_challenge? - @callback_response_body == challenge - end - - def subscribing? - mode == 'subscribe' - end - - def unsubscribing? - mode == 'unsubscribe' - end - - def callback_get_with_params - Request.new(:get, subscription.callback_url, params: callback_params).perform do |response| - @callback_response_body = response.body_with_limit - end - end - - def callback_params - { - 'hub.topic': account_url(subscription.account, format: :atom), - 'hub.mode': mode, - 'hub.challenge': challenge, - 'hub.lease_seconds': subscription.lease_seconds, - } - end - - def prepare_subscription - subscription.secret = secret - subscription.lease_seconds = lease_seconds - subscription.confirmed = true - end - - def challenge - @_challenge ||= SecureRandom.hex - end + def perform(subscription_id, mode, secret = nil, lease_seconds = nil); end end diff --git a/app/workers/pubsubhubbub/delivery_worker.rb b/app/workers/pubsubhubbub/delivery_worker.rb index 619bfa48a..1260060bd 100644 --- a/app/workers/pubsubhubbub/delivery_worker.rb +++ b/app/workers/pubsubhubbub/delivery_worker.rb @@ -2,80 +2,8 @@ class Pubsubhubbub::DeliveryWorker include Sidekiq::Worker - include RoutingHelper sidekiq_options queue: 'push', retry: 3, dead: false - sidekiq_retry_in do |count| - 5 * (count + 1) - end - - attr_reader :subscription, :payload - - def perform(subscription_id, payload) - @subscription = Subscription.find(subscription_id) - @payload = payload - process_delivery unless blocked_domain? - rescue => e - raise e.class, "Delivery failed for #{subscription&.callback_url}: #{e.message}", e.backtrace[0] - end - - private - - def process_delivery - callback_post_payload do |payload_delivery| - raise Mastodon::UnexpectedResponseError, payload_delivery unless response_successful? payload_delivery - end - - subscription.touch(:last_successful_delivery_at) - end - - def callback_post_payload(&block) - request = Request.new(:post, subscription.callback_url, body: payload) - request.add_headers(headers) - request.perform(&block) - end - - def blocked_domain? - DomainBlock.blocked?(host) - end - - def host - Addressable::URI.parse(subscription.callback_url).normalized_host - end - - def headers - { - 'Content-Type' => 'application/atom+xml', - 'Link' => link_header, - }.merge(signature_headers.to_h) - end - - def link_header - LinkHeader.new([hub_link_header, self_link_header]).to_s - end - - def hub_link_header - [api_push_url, [%w(rel hub)]] - end - - def self_link_header - [account_url(subscription.account, format: :atom), [%w(rel self)]] - end - - def signature_headers - { 'X-Hub-Signature' => payload_signature } if subscription.secret? - end - - def payload_signature - "sha1=#{hmac_payload_digest}" - end - - def hmac_payload_digest - OpenSSL::HMAC.hexdigest(OpenSSL::Digest.new('sha1'), subscription.secret, payload) - end - - def response_successful?(payload_delivery) - payload_delivery.code > 199 && payload_delivery.code < 300 - end + def perform(subscription_id, payload); end end diff --git a/app/workers/pubsubhubbub/distribution_worker.rb b/app/workers/pubsubhubbub/distribution_worker.rb index fed5e917d..75bac5d6f 100644 --- a/app/workers/pubsubhubbub/distribution_worker.rb +++ b/app/workers/pubsubhubbub/distribution_worker.rb @@ -5,28 +5,5 @@ class Pubsubhubbub::DistributionWorker sidekiq_options queue: 'push' - def perform(stream_entry_ids) - stream_entries = StreamEntry.where(id: stream_entry_ids).includes(:status).reject { |e| e.status.nil? || e.status.hidden? } - - return if stream_entries.empty? - - @account = stream_entries.first.account - @subscriptions = active_subscriptions.to_a - - distribute_public!(stream_entries) - end - - private - - def distribute_public!(stream_entries) - @payload = OStatus::AtomSerializer.render(OStatus::AtomSerializer.new.feed(@account, stream_entries)) - - Pubsubhubbub::DeliveryWorker.push_bulk(@subscriptions) do |subscription_id| - [subscription_id, @payload] - end - end - - def active_subscriptions - Subscription.where(account: @account).active.pluck(:id) - end + def perform(stream_entry_ids); end end diff --git a/app/workers/pubsubhubbub/raw_distribution_worker.rb b/app/workers/pubsubhubbub/raw_distribution_worker.rb index 16962a623..ece9c80ac 100644 --- a/app/workers/pubsubhubbub/raw_distribution_worker.rb +++ b/app/workers/pubsubhubbub/raw_distribution_worker.rb @@ -5,18 +5,5 @@ class Pubsubhubbub::RawDistributionWorker sidekiq_options queue: 'push' - def perform(xml, source_account_id) - @account = Account.find(source_account_id) - @subscriptions = active_subscriptions.to_a - - Pubsubhubbub::DeliveryWorker.push_bulk(@subscriptions) do |subscription| - [subscription.id, xml] - end - end - - private - - def active_subscriptions - Subscription.where(account: @account).active.select('id, callback_url, domain') - end + def perform(xml, source_account_id); end end diff --git a/app/workers/pubsubhubbub/subscribe_worker.rb b/app/workers/pubsubhubbub/subscribe_worker.rb index 2e176d1c1..b861b5e67 100644 --- a/app/workers/pubsubhubbub/subscribe_worker.rb +++ b/app/workers/pubsubhubbub/subscribe_worker.rb @@ -5,30 +5,5 @@ class Pubsubhubbub::SubscribeWorker sidekiq_options queue: 'push', retry: 10, unique: :until_executed, dead: false - sidekiq_retry_in do |count| - case count - when 0 - 30.minutes.seconds - when 1 - 2.hours.seconds - when 2 - 12.hours.seconds - else - 24.hours.seconds * (count - 2) - end - end - - sidekiq_retries_exhausted do |msg, _e| - account = Account.find(msg['args'].first) - Sidekiq.logger.error "PuSH subscription attempts for #{account.acct} exhausted. Unsubscribing" - ::UnsubscribeService.new.call(account) - end - - def perform(account_id) - account = Account.find(account_id) - logger.debug "PuSH re-subscribing to #{account.acct}" - ::SubscribeService.new.call(account) - rescue => e - raise e.class, "Subscribe failed for #{account&.acct}: #{e.message}", e.backtrace[0] - end + def perform(account_id); end end diff --git a/app/workers/pubsubhubbub/unsubscribe_worker.rb b/app/workers/pubsubhubbub/unsubscribe_worker.rb index a271715b7..0c1c263f6 100644 --- a/app/workers/pubsubhubbub/unsubscribe_worker.rb +++ b/app/workers/pubsubhubbub/unsubscribe_worker.rb @@ -5,11 +5,5 @@ class Pubsubhubbub::UnsubscribeWorker sidekiq_options queue: 'push', retry: false, unique: :until_executed, dead: false - def perform(account_id) - account = Account.find(account_id) - logger.debug "PuSH unsubscribing from #{account.acct}" - ::UnsubscribeService.new.call(account) - rescue ActiveRecord::RecordNotFound - true - end + def perform(account_id); end end diff --git a/app/workers/remote_profile_update_worker.rb b/app/workers/remote_profile_update_worker.rb index 03585ad2d..01e8daf8f 100644 --- a/app/workers/remote_profile_update_worker.rb +++ b/app/workers/remote_profile_update_worker.rb @@ -5,9 +5,5 @@ class RemoteProfileUpdateWorker sidekiq_options queue: 'pull' - def perform(account_id, body, resubscribe) - UpdateRemoteProfileService.new.call(body, Account.find(account_id), resubscribe) - rescue ActiveRecord::RecordNotFound - true - end + def perform(account_id, body, resubscribe); end end diff --git a/app/workers/salmon_worker.rb b/app/workers/salmon_worker.rb index d37d40432..10200b06c 100644 --- a/app/workers/salmon_worker.rb +++ b/app/workers/salmon_worker.rb @@ -5,9 +5,5 @@ class SalmonWorker sidekiq_options backtrace: true - def perform(account_id, body) - ProcessInteractionService.new.call(body, Account.find(account_id)) - rescue Nokogiri::XML::XPath::SyntaxError, ActiveRecord::RecordNotFound - true - end + def perform(account_id, body); end end diff --git a/app/workers/scheduler/preview_cards_cleanup_scheduler.rb b/app/workers/scheduler/preview_cards_cleanup_scheduler.rb new file mode 100644 index 000000000..2b38792f0 --- /dev/null +++ b/app/workers/scheduler/preview_cards_cleanup_scheduler.rb @@ -0,0 +1,22 @@ +# frozen_string_literal: true + +class Scheduler::PreviewCardsCleanupScheduler + include Sidekiq::Worker + + sidekiq_options unique: :until_executed, retry: 0 + + def perform + Maintenance::UncachePreviewWorker.push_bulk(recent_link_preview_cards.pluck(:id)) + Maintenance::UncachePreviewWorker.push_bulk(older_preview_cards.pluck(:id)) + end + + private + + def recent_link_preview_cards + PreviewCard.where(type: :link).where('updated_at < ?', 1.month.ago) + end + + def older_preview_cards + PreviewCard.where('updated_at < ?', 6.months.ago) + end +end diff --git a/app/workers/scheduler/subscriptions_scheduler.rb b/app/workers/scheduler/subscriptions_scheduler.rb index d5873bccb..6903cadc7 100644 --- a/app/workers/scheduler/subscriptions_scheduler.rb +++ b/app/workers/scheduler/subscriptions_scheduler.rb @@ -5,13 +5,5 @@ class Scheduler::SubscriptionsScheduler sidekiq_options unique: :until_executed, retry: 0 - def perform - Pubsubhubbub::SubscribeWorker.push_bulk(expiring_accounts.pluck(:id)) - end - - private - - def expiring_accounts - Account.expiring(1.day.from_now).partitioned - end + def perform; end end |