Change sidekiq-bulk's batch size from 10,000 to 1,000 jobs in one Red…

…is call (mastodon#24034)

app/lib/activitypub/forwarder.rb

@@ -12,7 +12,7 @@ def forwardable?
   end
 
   def forward!
-    ActivityPub::LowPriorityDeliveryWorker.push_bulk(inboxes) do |inbox_url|
+    ActivityPub::LowPriorityDeliveryWorker.push_bulk(inboxes, limit: 1_000) do |inbox_url|
       [payload, signature_account_id, inbox_url]
     end
   end

app/services/delete_account_service.rb

@@ -257,11 +257,11 @@ def purge_association(association_name)
   end
 
   def delete_actor!
-    ActivityPub::DeliveryWorker.push_bulk(delivery_inboxes) do |inbox_url|
+    ActivityPub::DeliveryWorker.push_bulk(delivery_inboxes, limit: 1_000) do |inbox_url|
       [delete_actor_json, @account.id, inbox_url]
     end
 
-    ActivityPub::LowPriorityDeliveryWorker.push_bulk(low_priority_delivery_inboxes) do |inbox_url|
+    ActivityPub::LowPriorityDeliveryWorker.push_bulk(low_priority_delivery_inboxes, limit: 1_000) do |inbox_url|
       [delete_actor_json, @account.id, inbox_url]
     end
   end

app/services/remove_status_service.rb

@@ -88,7 +88,7 @@ def remove_from_remote_reach
 
     status_reach_finder = StatusReachFinder.new(@status, unsafe: true)
 
-    ActivityPub::DeliveryWorker.push_bulk(status_reach_finder.inboxes) do |inbox_url|
+    ActivityPub::DeliveryWorker.push_bulk(status_reach_finder.inboxes, limit: 1_000) do |inbox_url|
       [signed_activity_json, @account.id, inbox_url]
     end
   end

app/services/suspend_account_service.rb

@@ -31,21 +31,21 @@ def reject_remote_follows!
     # counterpart to this operation, i.e. you can't then force a remote
     # account to re-follow you, so this part is not reversible.
 
-    follows = Follow.where(account: @account).to_a
+    Follow.where(account: @account).find_in_batches do |follows|
+      ActivityPub::DeliveryWorker.push_bulk(follows) do |follow|
+        [Oj.dump(serialize_payload(follow, ActivityPub::RejectFollowSerializer)), follow.target_account_id, @account.inbox_url]
+      end
 
-    ActivityPub::DeliveryWorker.push_bulk(follows) do |follow|
-      [Oj.dump(serialize_payload(follow, ActivityPub::RejectFollowSerializer)), follow.target_account_id, @account.inbox_url]
+      follows.each(&:destroy)
     end
-
-    follows.each(&:destroy)
   end
 
   def distribute_update_actor!
     return unless @account.local?
 
     account_reach_finder = AccountReachFinder.new(@account)
 
-    ActivityPub::DeliveryWorker.push_bulk(account_reach_finder.inboxes) do |inbox_url|
+    ActivityPub::DeliveryWorker.push_bulk(account_reach_finder.inboxes, limit: 1_000) do |inbox_url|
       [signed_activity_json, @account.id, inbox_url]
     end
   end

app/services/unsuspend_account_service.rb

@@ -41,7 +41,7 @@ def distribute_update_actor!
 
     account_reach_finder = AccountReachFinder.new(@account)
 
-    ActivityPub::DeliveryWorker.push_bulk(account_reach_finder.inboxes) do |inbox_url|
+    ActivityPub::DeliveryWorker.push_bulk(account_reach_finder.inboxes, limit: 1_000) do |inbox_url|
       [signed_activity_json, @account.id, inbox_url]
     end
   end

app/services/update_account_service.rb

@@ -22,7 +22,7 @@ def call(account, params, raise_error: false)
   def authorize_all_follow_requests(account)
     follow_requests = FollowRequest.where(target_account: account)
     follow_requests = follow_requests.preload(:account).select { |req| !req.account.silenced? }
-    AuthorizeFollowWorker.push_bulk(follow_requests) do |req|
+    AuthorizeFollowWorker.push_bulk(follow_requests, limit: 1_000) do |req|
       [req.account_id, req.target_account_id]
     end
   end

app/workers/activitypub/distribute_poll_update_worker.rb

@@ -12,7 +12,7 @@ def perform(status_id)
 
     return unless @status.preloadable_poll
 
-    ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
+    ActivityPub::DeliveryWorker.push_bulk(inboxes, limit: 1_000) do |inbox_url|
       [payload, @account.id, inbox_url]
     end
 

app/workers/activitypub/move_distribution_worker.rb

@@ -10,7 +10,7 @@ def perform(migration_id)
     @migration = AccountMigration.find(migration_id)
     @account   = @migration.account
 
-    ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
+    ActivityPub::DeliveryWorker.push_bulk(inboxes, limit: 1_000) do |inbox_url|
       [signed_payload, @account.id, inbox_url]
     end
 

app/workers/activitypub/raw_distribution_worker.rb

@@ -25,7 +25,7 @@ def perform(json, source_account_id, exclude_inboxes = [])
   def distribute!
     return if inboxes.empty?
 
-    ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
+    ActivityPub::DeliveryWorker.push_bulk(inboxes, limit: 1_000) do |inbox_url|
       [payload, source_account_id, inbox_url, options]
     end
   end

lib/cli.rb

@@ -131,7 +131,7 @@ def self_destruct
         json = Oj.dump(ActivityPub::LinkedDataSignature.new(payload).sign!(account))
 
         unless options[:dry_run]
-          ActivityPub::DeliveryWorker.push_bulk(inboxes) do |inbox_url|
+          ActivityPub::DeliveryWorker.push_bulk(inboxes, limit: 1_000) do |inbox_url|
             [json, account.id, inbox_url]
           end