From b7c1b12367b307d07303ce99f2c27bf255ecd56a Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Mon, 3 Apr 2017 18:55:06 +0200 Subject: Make default admin UI page reports. Add admin UI for creating a domain block --- app/services/block_domain_service.rb | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) (limited to 'app/services') diff --git a/app/services/block_domain_service.rb b/app/services/block_domain_service.rb index 9518b1fcf..6c131bd34 100644 --- a/app/services/block_domain_service.rb +++ b/app/services/block_domain_service.rb @@ -1,13 +1,11 @@ # frozen_string_literal: true class BlockDomainService < BaseService - def call(domain, severity) - DomainBlock.where(domain: domain).first_or_create!(domain: domain, severity: severity) - - if severity == :silence - Account.where(domain: domain).update_all(silenced: true) + def call(domain_block) + if domain_block.silence? + Account.where(domain: domain_block.domain).update_all(silenced: true) else - Account.where(domain: domain).find_each do |account| + Account.where(domain: domain_block.domain).find_each do |account| account.subscription(api_subscription_url(account.id)).unsubscribe if account.subscribed? SuspendAccountService.new.call(account) end -- cgit From 8232f76c482d3046055bd7bf224ef7835d0fa399 Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Mon, 3 Apr 2017 22:54:46 +0200 Subject: Add check for visibility.nil? even though it can't ever be, to check for race conditions --- app/lib/exceptions.rb | 1 + app/services/fan_out_on_write_service.rb | 2 ++ 2 files changed, 3 insertions(+) (limited to 'app/services') diff --git a/app/lib/exceptions.rb b/app/lib/exceptions.rb index 200da9fe1..9bc802c12 100644 --- a/app/lib/exceptions.rb +++ b/app/lib/exceptions.rb @@ -4,4 +4,5 @@ module Mastodon class Error < StandardError; end class NotPermittedError < Error; end class ValidationError < Error; end + class RaceConditionError < Error; end end diff --git a/app/services/fan_out_on_write_service.rb b/app/services/fan_out_on_write_service.rb index 402b84b2f..df404cbef 100644 --- a/app/services/fan_out_on_write_service.rb +++ b/app/services/fan_out_on_write_service.rb @@ -4,6 +4,8 @@ class FanOutOnWriteService < BaseService # Push a status into home and mentions feeds # @param [Status] status def call(status) + raise Mastodon::RaceConditionError if status.visibility.nil? + deliver_to_self(status) if status.account.local? if status.direct_visibility? -- cgit From 82aaedec467815c2947a11651d5216bb88ce4038 Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Tue, 4 Apr 2017 13:58:34 +0200 Subject: Reduce number of items in feeds, optimize regeneration worker slightly, make regeneration worker unique, (only schedule/execute once at a time) --- Gemfile | 2 ++ Gemfile.lock | 9 +++++++++ app/lib/feed_manager.rb | 6 +++--- app/services/precompute_feed_service.rb | 8 +++++--- app/workers/regeneration_worker.rb | 2 +- 5 files changed, 20 insertions(+), 7 deletions(-) (limited to 'app/services') diff --git a/Gemfile b/Gemfile index cb9824131..41c636904 100644 --- a/Gemfile +++ b/Gemfile @@ -46,6 +46,8 @@ gem 'will_paginate' gem 'rack-attack' gem 'rack-cors', require: 'rack/cors' gem 'sidekiq' +gem 'sidekiq-unique-jobs' +gem 'sidekiq-merger' gem 'rails-settings-cached' gem 'simple-navigation' gem 'statsd-instrument' diff --git a/Gemfile.lock b/Gemfile.lock index 6e3115249..27de1bee0 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -387,6 +387,13 @@ GEM connection_pool (~> 2.2, >= 2.2.0) rack-protection (>= 1.5.0) redis (~> 3.2, >= 3.2.1) + sidekiq-merger (0.0.11) + activesupport (>= 3.2, < 6) + concurrent-ruby (~> 1.0) + sidekiq (>= 3.4, < 5) + sidekiq-unique-jobs (4.0.18) + sidekiq (>= 2.6) + thor simple-navigation (4.0.3) activesupport (>= 2.3.2) simple_form (3.2.1) @@ -510,6 +517,8 @@ DEPENDENCIES sass-rails (~> 5.0) sdoc (~> 0.4.0) sidekiq + sidekiq-merger + sidekiq-unique-jobs simple-navigation simple_form simplecov diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index 919bc3df9..a2efcce10 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -5,7 +5,7 @@ require 'singleton' class FeedManager include Singleton - MAX_ITEMS = 800 + MAX_ITEMS = 400 def key(type, id) "feed:#{type}:#{id}" @@ -50,9 +50,9 @@ class FeedManager def merge_into_timeline(from_account, into_account) timeline_key = key(:home, into_account.id) - query = from_account.statuses.limit(MAX_ITEMS) + query = from_account.statuses.limit(FeedManager::MAX_ITEMS / 4) - if redis.zcard(timeline_key) >= FeedManager::MAX_ITEMS + if redis.zcard(timeline_key) >= FeedManager::MAX_ITEMS / 4 oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0 query = query.where('id > ?', oldest_home_score) end diff --git a/app/services/precompute_feed_service.rb b/app/services/precompute_feed_service.rb index e1ec56e8d..a57c401d0 100644 --- a/app/services/precompute_feed_service.rb +++ b/app/services/precompute_feed_service.rb @@ -5,9 +5,11 @@ class PrecomputeFeedService < BaseService # @param [Symbol] type :home or :mentions # @param [Account] account def call(_, account) - Status.as_home_timeline(account).limit(FeedManager::MAX_ITEMS).each do |status| - next if status.direct_visibility? || FeedManager.instance.filter?(:home, status, account) - redis.zadd(FeedManager.instance.key(:home, account.id), status.id, status.reblog? ? status.reblog_of_id : status.id) + redis.pipelined do + Status.as_home_timeline(account).limit(FeedManager::MAX_ITEMS / 4).each do |status| + next if status.direct_visibility? || FeedManager.instance.filter?(:home, status, account) + redis.zadd(FeedManager.instance.key(:home, account.id), status.id, status.reblog? ? status.reblog_of_id : status.id) + end end end diff --git a/app/workers/regeneration_worker.rb b/app/workers/regeneration_worker.rb index 82665b581..da8b845f6 100644 --- a/app/workers/regeneration_worker.rb +++ b/app/workers/regeneration_worker.rb @@ -3,7 +3,7 @@ class RegenerationWorker include Sidekiq::Worker - sidekiq_options queue: 'pull', backtrace: true + sidekiq_options queue: 'pull', backtrace: true, unique: :until_executed def perform(account_id, _ = :home) PrecomputeFeedService.new.call(:home, Account.find(account_id)) -- cgit From 6fd865c0004efbf11ee87c06fea9f48af567fabe Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Tue, 4 Apr 2017 19:21:37 +0200 Subject: Spawn FeedInsertWorker to deliver status into personal feed --- app/lib/feed_manager.rb | 32 ++++++++++++++++---------------- app/services/fan_out_on_write_service.rb | 13 ++++++------- app/services/notify_service.rb | 2 +- app/services/precompute_feed_service.rb | 2 +- app/workers/feed_insert_worker.rb | 15 +++++++++++++++ 5 files changed, 39 insertions(+), 25 deletions(-) create mode 100644 app/workers/feed_insert_worker.rb (limited to 'app/services') diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index a2efcce10..28e712704 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -11,11 +11,11 @@ class FeedManager "feed:#{type}:#{id}" end - def filter?(timeline_type, status, receiver) + def filter?(timeline_type, status, receiver_id) if timeline_type == :home - filter_from_home?(status, receiver) + filter_from_home?(status, receiver_id) elsif timeline_type == :mentions - filter_from_mentions?(status, receiver) + filter_from_mentions?(status, receiver_id) else false end @@ -91,39 +91,39 @@ class FeedManager Redis.current end - def filter_from_home?(status, receiver) + def filter_from_home?(status, receiver_id) return true if status.reply? && status.in_reply_to_id.nil? check_for_mutes = [status.account_id] check_for_mutes.concat([status.reblog.account_id]) if status.reblog? - return true if receiver.muting?(check_for_mutes) + return true if Mute.where(account_id: receiver_id, target_account_id: check_for_mutes).any? check_for_blocks = status.mentions.map(&:account_id) check_for_blocks.concat([status.reblog.account_id]) if status.reblog? - return true if receiver.blocking?(check_for_blocks) + return true if Block.where(account_id: receiver_id, target_account_id: check_for_blocks).any? - if status.reply? && !status.in_reply_to_account_id.nil? # Filter out if it's a reply - should_filter = !receiver.following?(status.in_reply_to_account) # and I'm not following the person it's a reply to - should_filter &&= !(receiver.id == status.in_reply_to_account_id) # and it's not a reply to me - should_filter &&= !(status.account_id == status.in_reply_to_account_id) # and it's not a self-reply + if status.reply? && !status.in_reply_to_account_id.nil? # Filter out if it's a reply + should_filter = !Follow.where(account_id: receiver_id, target_account_id: status.in_reply_to_account_id).exists? # and I'm not following the person it's a reply to + should_filter &&= !(receiver_id == status.in_reply_to_account_id) # and it's not a reply to me + should_filter &&= !(status.account_id == status.in_reply_to_account_id) # and it's not a self-reply return should_filter - elsif status.reblog? # Filter out a reblog - return status.reblog.account.blocking?(receiver) # or if the author of the reblogged status is blocking me + elsif status.reblog? # Filter out a reblog + return Block.where(account_id: status.reblog.account_id, target_account_id: receiver_id).exists? # or if the author of the reblogged status is blocking me end false end - def filter_from_mentions?(status, receiver) + def filter_from_mentions?(status, receiver_id) check_for_blocks = [status.account_id] check_for_blocks.concat(status.mentions.select('account_id').map(&:account_id)) check_for_blocks.concat([status.in_reply_to_account]) if status.reply? && !status.in_reply_to_account_id.nil? - should_filter = receiver.id == status.account_id # Filter if I'm mentioning myself - should_filter ||= receiver.blocking?(check_for_blocks) # or it's from someone I blocked, in reply to someone I blocked, or mentioning someone I blocked - should_filter ||= (status.account.silenced? && !receiver.following?(status.account)) # of if the account is silenced and I'm not following them + should_filter = receiver_id == status.account_id # Filter if I'm mentioning myself + should_filter ||= Block.where(account_id: receiver_id, target_account_id: check_for_blocks).any? # or it's from someone I blocked, in reply to someone I blocked, or mentioning someone I blocked + should_filter ||= (status.account.silenced? && !Follow.where(account_id: receiver_id, target_account_id: status.account_id).exists?) # of if the account is silenced and I'm not following them should_filter end diff --git a/app/services/fan_out_on_write_service.rb b/app/services/fan_out_on_write_service.rb index df404cbef..42222c25b 100644 --- a/app/services/fan_out_on_write_service.rb +++ b/app/services/fan_out_on_write_service.rb @@ -33,9 +33,8 @@ class FanOutOnWriteService < BaseService def deliver_to_followers(status) Rails.logger.debug "Delivering status #{status.id} to followers" - status.account.followers.where(domain: nil).joins(:user).where('users.current_sign_in_at > ?', 14.days.ago).find_each do |follower| - next if FeedManager.instance.filter?(:home, status, follower) - FeedManager.instance.push(:home, follower, status) + status.account.followers.where(domain: nil).joins(:user).where('users.current_sign_in_at > ?', 14.days.ago).select(:id).find_each do |follower| + FeedInsertWorker.perform_async(status.id, follower.id) end end @@ -44,7 +43,7 @@ class FanOutOnWriteService < BaseService status.mentions.includes(:account).each do |mention| mentioned_account = mention.account - next if !mentioned_account.local? || !mentioned_account.following?(status.account) || FeedManager.instance.filter?(:home, status, mentioned_account) + next if !mentioned_account.local? || !mentioned_account.following?(status.account) || FeedManager.instance.filter?(:home, status, mention.account_id) FeedManager.instance.push(:home, mentioned_account, status) end end @@ -54,9 +53,9 @@ class FanOutOnWriteService < BaseService payload = FeedManager.instance.inline_render(nil, 'api/v1/statuses/show', status) - status.tags.find_each do |tag| - FeedManager.instance.broadcast("hashtag:#{tag.name}", event: 'update', payload: payload) - FeedManager.instance.broadcast("hashtag:#{tag.name}:local", event: 'update', payload: payload) if status.account.local? + status.tags.pluck(:name).each do |hashtag| + FeedManager.instance.broadcast("hashtag:#{hashtag}", event: 'update', payload: payload) + FeedManager.instance.broadcast("hashtag:#{hashtag}:local", event: 'update', payload: payload) if status.account.local? end end diff --git a/app/services/notify_service.rb b/app/services/notify_service.rb index 942cd9d21..24486f220 100644 --- a/app/services/notify_service.rb +++ b/app/services/notify_service.rb @@ -17,7 +17,7 @@ class NotifyService < BaseService private def blocked_mention? - FeedManager.instance.filter?(:mentions, @notification.mention.status, @recipient) + FeedManager.instance.filter?(:mentions, @notification.mention.status, @recipient.id) end def blocked_favourite? diff --git a/app/services/precompute_feed_service.rb b/app/services/precompute_feed_service.rb index a57c401d0..07dcb81da 100644 --- a/app/services/precompute_feed_service.rb +++ b/app/services/precompute_feed_service.rb @@ -7,7 +7,7 @@ class PrecomputeFeedService < BaseService def call(_, account) redis.pipelined do Status.as_home_timeline(account).limit(FeedManager::MAX_ITEMS / 4).each do |status| - next if status.direct_visibility? || FeedManager.instance.filter?(:home, status, account) + next if status.direct_visibility? || FeedManager.instance.filter?(:home, status, account.id) redis.zadd(FeedManager.instance.key(:home, account.id), status.id, status.reblog? ? status.reblog_of_id : status.id) end end diff --git a/app/workers/feed_insert_worker.rb b/app/workers/feed_insert_worker.rb new file mode 100644 index 000000000..a58dfaa74 --- /dev/null +++ b/app/workers/feed_insert_worker.rb @@ -0,0 +1,15 @@ +# frozen_string_literal: true + +class FeedInsertWorker + include Sidekiq::Worker + + def perform(status_id, follower_id) + status = Status.find(status_id) + follower = Account.find(follower_id) + + return if FeedManager.instance.filter?(:home, status, follower.id) + FeedManager.instance.push(:home, follower, status) + rescue ActiveRecord::RecordNotFound + true + end +end -- cgit From 220bc48e8e9c4b8cebd98537233998f34d768347 Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Wed, 5 Apr 2017 14:26:17 +0200 Subject: Only render public payload once in FanOutOnWrite --- app/services/fan_out_on_write_service.rb | 17 +++++++++-------- app/workers/after_remote_follow_request_worker.rb | 2 ++ app/workers/after_remote_follow_worker.rb | 2 ++ 3 files changed, 13 insertions(+), 8 deletions(-) (limited to 'app/services') diff --git a/app/services/fan_out_on_write_service.rb b/app/services/fan_out_on_write_service.rb index 42222c25b..106d257ba 100644 --- a/app/services/fan_out_on_write_service.rb +++ b/app/services/fan_out_on_write_service.rb @@ -16,6 +16,7 @@ class FanOutOnWriteService < BaseService return if status.account.silenced? || !status.public_visibility? || status.reblog? + render_anonymous_payload(status) deliver_to_hashtags(status) return if status.reply? && status.in_reply_to_account_id != status.account_id @@ -48,23 +49,23 @@ class FanOutOnWriteService < BaseService end end + def render_anonymous_payload(status) + @payload = FeedManager.instance.inline_render(nil, 'api/v1/statuses/show', status) + end + def deliver_to_hashtags(status) Rails.logger.debug "Delivering status #{status.id} to hashtags" - payload = FeedManager.instance.inline_render(nil, 'api/v1/statuses/show', status) - status.tags.pluck(:name).each do |hashtag| - FeedManager.instance.broadcast("hashtag:#{hashtag}", event: 'update', payload: payload) - FeedManager.instance.broadcast("hashtag:#{hashtag}:local", event: 'update', payload: payload) if status.account.local? + FeedManager.instance.broadcast("hashtag:#{hashtag}", event: 'update', payload: @payload) + FeedManager.instance.broadcast("hashtag:#{hashtag}:local", event: 'update', payload: @payload) if status.account.local? end end def deliver_to_public(status) Rails.logger.debug "Delivering status #{status.id} to public timeline" - payload = FeedManager.instance.inline_render(nil, 'api/v1/statuses/show', status) - - FeedManager.instance.broadcast(:public, event: 'update', payload: payload) - FeedManager.instance.broadcast('public:local', event: 'update', payload: payload) if status.account.local? + FeedManager.instance.broadcast(:public, event: 'update', payload: @payload) + FeedManager.instance.broadcast('public:local', event: 'update', payload: @payload) if status.account.local? end end diff --git a/app/workers/after_remote_follow_request_worker.rb b/app/workers/after_remote_follow_request_worker.rb index 1f2db3061..928069211 100644 --- a/app/workers/after_remote_follow_request_worker.rb +++ b/app/workers/after_remote_follow_request_worker.rb @@ -13,5 +13,7 @@ class AfterRemoteFollowRequestWorker follow_request.destroy FollowService.new.call(follow_request.account, updated_account.acct) + rescue ActiveRecord::RecordNotFound + true end end diff --git a/app/workers/after_remote_follow_worker.rb b/app/workers/after_remote_follow_worker.rb index bdd2c2a91..d12fa3454 100644 --- a/app/workers/after_remote_follow_worker.rb +++ b/app/workers/after_remote_follow_worker.rb @@ -13,5 +13,7 @@ class AfterRemoteFollowWorker follow.destroy FollowService.new.call(follow.account, updated_account.acct) + rescue ActiveRecord::RecordNotFound + true end end -- cgit From 5b95be1c42ba69c9a3a79cfa990c80a5f2debfc6 Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Wed, 5 Apr 2017 19:45:18 +0200 Subject: Replace calls to FeedManager#inline_render and #broadcast --- app/lib/feed_manager.rb | 2 +- app/lib/inline_renderer.rb | 13 +++++++++++++ app/services/fan_out_on_write_service.rb | 10 +++++----- app/services/notify_service.rb | 2 +- app/services/remove_status_service.rb | 6 +++--- app/workers/push_update_worker.rb | 13 +++---------- 6 files changed, 26 insertions(+), 20 deletions(-) create mode 100644 app/lib/inline_renderer.rb (limited to 'app/services') diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index 87865bfdc..58d9fb1fc 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -34,7 +34,7 @@ class FeedManager trim(timeline_type, account.id) end - PushUpdateWorker.perform_async(timeline_type, account.id, status.id) + PushUpdateWorker.perform_async(account.id, status.id) end def trim(type, account_id) diff --git a/app/lib/inline_renderer.rb b/app/lib/inline_renderer.rb new file mode 100644 index 000000000..8e04ad1d5 --- /dev/null +++ b/app/lib/inline_renderer.rb @@ -0,0 +1,13 @@ +# frozen_string_literal: true + +class InlineRenderer + def self.render(status, current_account, template) + Rabl::Renderer.new( + template, + status, + view_path: 'app/views', + format: :json, + scope: InlineRablScope.new(current_account) + ).render + end +end diff --git a/app/services/fan_out_on_write_service.rb b/app/services/fan_out_on_write_service.rb index 106d257ba..c63fcc1fe 100644 --- a/app/services/fan_out_on_write_service.rb +++ b/app/services/fan_out_on_write_service.rb @@ -50,22 +50,22 @@ class FanOutOnWriteService < BaseService end def render_anonymous_payload(status) - @payload = FeedManager.instance.inline_render(nil, 'api/v1/statuses/show', status) + @payload = InlineRenderer.render(status, nil, 'api/v1/statuses/show') end def deliver_to_hashtags(status) Rails.logger.debug "Delivering status #{status.id} to hashtags" status.tags.pluck(:name).each do |hashtag| - FeedManager.instance.broadcast("hashtag:#{hashtag}", event: 'update', payload: @payload) - FeedManager.instance.broadcast("hashtag:#{hashtag}:local", event: 'update', payload: @payload) if status.account.local? + Redis.current.publish("hashtag:#{hashtag}", Oj.dump(event: :update, payload: @payload)) + Redis.current.publish("hashtag:#{hashtag}:local", Oj.dump(event: :update, payload: @payload)) if status.account.local? end end def deliver_to_public(status) Rails.logger.debug "Delivering status #{status.id} to public timeline" - FeedManager.instance.broadcast(:public, event: 'update', payload: @payload) - FeedManager.instance.broadcast('public:local', event: 'update', payload: @payload) if status.account.local? + Redis.current.publish('public', Oj.dump(event: 'update', payload: @payload)) + Redis.current.publish('public:local', Oj.dump(event: 'update', payload: @payload)) if status.account.local? end end diff --git a/app/services/notify_service.rb b/app/services/notify_service.rb index 24486f220..62508a049 100644 --- a/app/services/notify_service.rb +++ b/app/services/notify_service.rb @@ -50,7 +50,7 @@ class NotifyService < BaseService def create_notification @notification.save! return unless @notification.browserable? - FeedManager.instance.broadcast(@recipient.id, event: 'notification', payload: FeedManager.instance.inline_render(@recipient, 'api/v1/notifications/show', @notification)) + Redis.current.publish(@recipient.id, Oj.dump(event: :notification, payload: InlineRenderer.render(@notification, @recipient, 'api/v1/notifications/show'))) end def send_email diff --git a/app/services/remove_status_service.rb b/app/services/remove_status_service.rb index cf1f432e4..e19fdd030 100644 --- a/app/services/remove_status_service.rb +++ b/app/services/remove_status_service.rb @@ -65,17 +65,17 @@ class RemoveStatusService < BaseService redis.zremrangebyscore(FeedManager.instance.key(type, receiver.id), status.id, status.id) end - FeedManager.instance.broadcast(receiver.id, event: 'delete', payload: status.id) + Redis.current.publish(receiver.id, Oj.dump(event: :delete, payload: status.id)) end def remove_from_hashtags(status) status.tags.each do |tag| - FeedManager.instance.broadcast("hashtag:#{tag.name}", event: 'delete', payload: status.id) + Redis.current.publish("hashtag:#{tag.name}", Oj.dump(event: :delete, payload: status.id)) end end def remove_from_public(status) - FeedManager.instance.broadcast(:public, event: 'delete', payload: status.id) + Redis.current.publish('public', Oj.dump(event: :delete, payload: status.id)) end def redis diff --git a/app/workers/push_update_worker.rb b/app/workers/push_update_worker.rb index 166a9b449..fbcdcf634 100644 --- a/app/workers/push_update_worker.rb +++ b/app/workers/push_update_worker.rb @@ -3,19 +3,12 @@ class PushUpdateWorker include Sidekiq::Worker - def perform(timeline, account_id, status_id) + def perform(account_id, status_id) account = Account.find(account_id) status = Status.find(status_id) - - message = Rabl::Renderer.new( - 'api/v1/statuses/show', - status, - view_path: 'app/views', - format: :json, - scope: InlineRablScope.new(account) - ) + message = InlineRenderer.render(status, account, 'api/v1/statuses/show') - Redis.current.publish("timeline:#{account.id}", Oj.dump({ event: :update, payload: message, queued_at: (Time.now.to_f * 1000.0).to_i })) + Redis.current.publish("timeline:#{account.id}", Oj.dump(event: :update, payload: message, queued_at: (Time.now.to_f * 1000.0).to_i)) rescue ActiveRecord::RecordNotFound true end -- cgit