DEV: reduce amount of errors logged when notifying on flags (#20472)

Forcing distributed muted to raise when a notify reviewable job is running
leads to excessive errors in the logs under many conditions.

The new pattern

1. Optimises the counting of reviewables so it is a lot faster
2. Holds the distributed lock for 2 minutes (max)


The downside is the job queue can get blocked up when tons of notify
reviewables are running at the same time. However this should be very
rare in the real world, as we only notify when stuff is flagged which
is fairly infrequent.

This also give a fair bit more time for the notifications which may be
a little slow on large sites with tons of mods.

app/jobs/regular/notify_reviewable.rb

@@ -22,14 +22,18 @@ class Jobs::NotifyReviewable < ::Jobs::Base
         end
     end
 
-    DistributedMutex.synchronize("notify_reviewable_job", validity: 10, max_get_lock_attempts: 1) do
+    DistributedMutex.synchronize("notify_reviewable_job", validity: 120) do
       counts = Hash.new(0)
-
-      Reviewable.default_visible.pending.each do |r|
-        counts[:admins] += 1
-        counts[:moderators] += 1 if r.reviewable_by_moderator?
-        counts[r.reviewable_by_group_id] += 1 if r.reviewable_by_group_id
-      end
+      Reviewable
+        .default_visible
+        .pending
+        .group(:reviewable_by_moderator, :reviewable_by_group_id)
+        .pluck(:reviewable_by_moderator, :reviewable_by_group_id, "count(*)")
+        .each do |reviewable_by_moderator, reviewable_by_group_id, count|
+          counts[:admins] += count
+          counts[:moderators] += count if reviewable_by_moderator
+          counts[reviewable_by_group_id] += count if reviewable_by_group_id
+        end
 
       if SiteSetting.legacy_navigation_menu?
         notify_legacy(

lib/distributed_mutex.rb

@@ -30,28 +30,16 @@ class DistributedMutex
     end
   LUA
 
-  def self.synchronize(
-    key,
-    redis: nil,
-    validity: DEFAULT_VALIDITY,
-    max_get_lock_attempts: nil,
-    &blk
-  )
-    self.new(
-      key,
-      redis: redis,
-      validity: validity,
-      max_get_lock_attempts: max_get_lock_attempts,
-    ).synchronize(&blk)
+  def self.synchronize(key, redis: nil, validity: DEFAULT_VALIDITY, &blk)
+    self.new(key, redis: redis, validity: validity).synchronize(&blk)
   end
 
-  def initialize(key, redis: nil, validity: DEFAULT_VALIDITY, max_get_lock_attempts: nil)
+  def initialize(key, redis: nil, validity: DEFAULT_VALIDITY)
     @key = key
     @using_global_redis = true if !redis
     @redis = redis || Discourse.redis
     @mutex = Mutex.new
     @validity = validity
-    @max_get_lock_attempts = max_get_lock_attempts
   end
 
   # NOTE wrapped in mutex to maintain its semantics
@@ -81,15 +69,11 @@ class DistributedMutex
     result
   end
 
-  class MaximumAttemptsExceeded < StandardError
-  end
-
   private
 
   attr_reader :key
   attr_reader :redis
   attr_reader :validity
-  attr_reader :max_get_lock_attempts
 
   def get_lock
     attempts = 0
@@ -108,10 +92,6 @@ class DistributedMutex
       if @using_global_redis && Discourse.recently_readonly? && attempts > CHECK_READONLY_ATTEMPTS
         raise Discourse::ReadOnly
       end
-
-      if max_get_lock_attempts && attempts > max_get_lock_attempts
-        raise DistributedMutex::MaximumAttemptsExceeded
-      end
     end
   end