Merge branch 'main' into glitch-soc/merge-upstream

app/workers/activitypub/processing_worker.rb

@@ -6,7 +6,10 @@ class ActivityPub::ProcessingWorker
   sidekiq_options backtrace: true, retry: 8
 
   def perform(account_id, body, delivered_to_account_id = nil)
-    ActivityPub::ProcessCollectionService.new.call(body, Account.find(account_id), override_timestamps: true, delivered_to_account_id: delivered_to_account_id, delivery: true)
+    account = Account.find_by(id: account_id)
+    return if account.nil?
+
+    ActivityPub::ProcessCollectionService.new.call(body, account, override_timestamps: true, delivered_to_account_id: delivered_to_account_id, delivery: true)
   rescue ActiveRecord::RecordInvalid => e
     Rails.logger.debug "Error processing incoming ActivityPub object: #{e}"
   end

app/workers/scheduler/accounts_statuses_cleanup_scheduler.rb

@@ -66,7 +66,7 @@ class Scheduler::AccountsStatusesCleanupScheduler
   end
 
   def compute_budget
-    threads = Sidekiq::ProcessSet.new.filter { |x| x['queues'].include?('push') }.map { |x| x['concurrency'] }.sum
+    threads = Sidekiq::ProcessSet.new.select { |x| x['queues'].include?('push') }.map { |x| x['concurrency'] }.sum
     [PER_THREAD_BUDGET * threads, MAX_BUDGET].min
   end
 

app/workers/scheduler/follow_recommendations_scheduler.rb

@@ -16,28 +16,33 @@ class Scheduler::FollowRecommendationsScheduler
     AccountSummary.refresh
     FollowRecommendation.refresh
 
-    fallback_recommendations = FollowRecommendation.order(rank: :desc).limit(SET_SIZE).index_by(&:account_id)
+    fallback_recommendations = FollowRecommendation.order(rank: :desc).limit(SET_SIZE)
 
-    I18n.available_locales.each do |locale|
+    I18n.available_locales.map { |locale| locale.to_s.split(/[_-]/).first }.uniq.each do |locale|
       recommendations = begin
         if AccountSummary.safe.filtered.localized(locale).exists? # We can skip the work if no accounts with that language exist
-          FollowRecommendation.localized(locale).order(rank: :desc).limit(SET_SIZE).index_by(&:account_id)
+          FollowRecommendation.localized(locale).order(rank: :desc).limit(SET_SIZE).map { |recommendation| [recommendation.account_id, recommendation.rank] }
         else
-          {}
+          []
         end
       end
 
       # Use language-agnostic results if there are not enough language-specific ones
-      missing = SET_SIZE - recommendations.keys.size
+      missing = SET_SIZE - recommendations.size
+
+      if missing.positive? && fallback_recommendations.size.positive?
+        max_fallback_rank = fallback_recommendations.first.rank || 0
+
+        # Language-specific results should be above language-agnostic ones,
+        # otherwise language-agnostic ones will always overshadow them
+        recommendations.map! { |(account_id, rank)| [account_id, rank + max_fallback_rank] }
 
-      if missing.positive?
         added = 0
 
-        # Avoid duplicate results
-        fallback_recommendations.each_value do |recommendation|
-          next if recommendations.key?(recommendation.account_id)
+        fallback_recommendations.each do |recommendation|
+          next if recommendations.any? { |(account_id, _)| account_id == recommendation.account_id }
 
-          recommendations[recommendation.account_id] = recommendation
+          recommendations << [recommendation.account_id, recommendation.rank]
           added += 1
 
           break if added >= missing
@@ -47,8 +52,8 @@ class Scheduler::FollowRecommendationsScheduler
       redis.pipelined do
         redis.del(key(locale))
 
-        recommendations.each_value do |recommendation|
-          redis.zadd(key(locale), recommendation.rank, recommendation.account_id)
+        recommendations.each do |(account_id, rank)|
+          redis.zadd(key(locale), rank, account_id)
         end
       end
     end