Prevent UserCleanupScheduler from overwhelming streaming (#25519)

app/services/remove_status_service.rb

@@ -12,6 +12,7 @@ class RemoveStatusService < BaseService
   # @option  [Boolean] :immediate
   # @option  [Boolean] :preserve
   # @option  [Boolean] :original_removed
+  # @option  [Boolean] :skip_streaming
   def call(status, **options)
     @payload  = Oj.dump(event: :delete, payload: status.id.to_s)
     @status   = status
@@ -52,6 +53,9 @@ class RemoveStatusService < BaseService
 
   private
 
+  # The following FeedManager calls all do not result in redis publishes for
+  # streaming, as the `:update` option is false
+
   def remove_from_self
     FeedManager.instance.unpush_from_home(@account, @status)
   end
@@ -75,6 +79,8 @@ class RemoveStatusService < BaseService
     # followers. Here we send a delete to actively mentioned accounts
     # that may not follow the account
 
+    return if skip_streaming?
+
     @status.active_mentions.find_each do |mention|
       redis.publish("timeline:#{mention.account_id}", @payload)
     end
@@ -103,7 +109,7 @@ class RemoveStatusService < BaseService
     # without us being able to do all the fancy stuff
 
     @status.reblogs.rewhere(deleted_at: [nil, @status.deleted_at]).includes(:account).reorder(nil).find_each do |reblog|
-      RemoveStatusService.new.call(reblog, original_removed: true)
+      RemoveStatusService.new.call(reblog, original_removed: true, skip_streaming: skip_streaming?)
     end
   end
 
@@ -114,6 +120,8 @@ class RemoveStatusService < BaseService
 
     return unless @status.public_visibility?
 
+    return if skip_streaming?
+
     @status.tags.map(&:name).each do |hashtag|
       redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}", @payload)
       redis.publish("timeline:hashtag:#{hashtag.mb_chars.downcase}:local", @payload) if @status.local?
@@ -123,6 +131,8 @@ class RemoveStatusService < BaseService
   def remove_from_public
     return unless @status.public_visibility?
 
+    return if skip_streaming?
+
     redis.publish('timeline:public', @payload)
     redis.publish(@status.local? ? 'timeline:public:local' : 'timeline:public:remote', @payload)
   end
@@ -130,6 +140,8 @@ class RemoveStatusService < BaseService
   def remove_from_media
     return unless @status.public_visibility?
 
+    return if skip_streaming?
+
     redis.publish('timeline:public:media', @payload)
     redis.publish(@status.local? ? 'timeline:public:local:media' : 'timeline:public:remote:media', @payload)
   end
@@ -143,4 +155,8 @@ class RemoveStatusService < BaseService
   def permanently?
     @options[:immediate] || !(@options[:preserve] || @status.reported?)
   end
+
+  def skip_streaming?
+    !!@options[:skip_streaming]
+  end
 end

app/workers/scheduler/user_cleanup_scheduler.rb

@@ -24,7 +24,7 @@ class Scheduler::UserCleanupScheduler
   def clean_discarded_statuses!
     Status.unscoped.discarded.where('deleted_at <= ?', 30.days.ago).find_in_batches do |statuses|
       RemovalWorker.push_bulk(statuses) do |status|
-        [status.id, { 'immediate' => true }]
+        [status.id, { 'immediate' => true, 'skip_streaming' => true }]
       end
     end
   end