PERF: Move per-post language detection job into batches using redis

Currently, upon post serialization, we spawn a
new job to detect the language of the post. This
could potentially flood sidekiq on larger sites.

This PR uses redis to keep track of posts that
need language detection, then schedules them in
a 5-minute-ly job, with a maximum of 1000 posts
at a time.

Author: nattsw

plugin.rb

@@ -17,6 +17,7 @@ module ::DiscourseTranslator
     PLUGIN_NAME = "discourse_translator".freeze
     DETECTED_LANG_CUSTOM_FIELD = "post_detected_lang".freeze
     TRANSLATED_CUSTOM_FIELD = "translated_text".freeze
+    LANG_DETECT_NEEDED = "lang_detect_needed".freeze
 
     autoload :Microsoft,
              "#{Rails.root}/plugins/discourse-translator/services/discourse_translator/microsoft"
@@ -134,6 +135,45 @@ def execute(args)
         end
       end
     end
+
+    class DetectPostsTranslation < ::Jobs::Scheduled
+      sidekiq_options retry: false
+      every 5.minutes
+
+      BATCH_SIZE = 100
+      MAX_QUEUE_SIZE = 1000
+
+      def execute(args)
+        return unless SiteSetting.translator_enabled
+
+        post_ids = Discourse.redis.spop(DiscourseTranslator::LANG_DETECT_NEEDED, MAX_QUEUE_SIZE)
+        return if post_ids.blank?
+
+        post_ids.each_slice(BATCH_SIZE) do |batch|
+          process_batch(batch)
+        end
+      end
+
+      private
+
+      def process_batch(post_ids)
+        posts = Post.where(id: post_ids).to_a
+        posts.each do |post|
+          DistributedMutex.synchronize("detect_translation_#{post.id}") do
+            begin
+              translator = "DiscourseTranslator::#{SiteSetting.translator}".constantize
+              translator.detect(post)
+              if !post.custom_fields_clean?
+                post.save_custom_fields
+                post.publish_change_to_clients!(:revised)
+              end
+            rescue ::DiscourseTranslator::ProblemCheckedTranslationError
+              # problem-checked translation errors gracefully
+            end
+          end
+        end
+      end
+    end
   end
 
   on(:post_process) { |post| Jobs.enqueue(:detect_translation, post_id: post.id) }
@@ -160,7 +200,7 @@ def execute(args)
     detected_lang = post_custom_fields[::DiscourseTranslator::DETECTED_LANG_CUSTOM_FIELD]
 
     if !detected_lang
-      Jobs.enqueue(:detect_translation, post_id: object.id)
+      Discourse.redis.sadd?(DiscourseTranslator::LANG_DETECT_NEEDED, object.id)
       false
     else
       detected_lang !=

spec/jobs/detect_posts_translation_spec.rb

@@ -0,0 +1,86 @@
+# frozen_string_literal: true
+
+require "aws-sdk-translate"
+
+describe Jobs::DetectPostsTranslation do
+  fab!(:posts) { Fabricate.times(5, :post) }
+
+  before do
+    SiteSetting.translator_enabled = true
+    SiteSetting.translator = "Amazon"
+    client = Aws::Translate::Client.new(stub_responses: true)
+    client.stub_responses(
+      :translate_text,
+      { translated_text: "大丈夫", source_language_code: "en", target_language_code: "jp" },
+    )
+    Aws::Translate::Client.stubs(:new).returns(client)
+    posts.each do |post|
+      Discourse.redis.sadd?(DiscourseTranslator::LANG_DETECT_NEEDED, post.id)
+    end
+  end
+
+  it "processes posts in batches and updates their translations" do
+    described_class.new.execute({})
+
+    posts.each do |post|
+      post.reload
+      expect(post.custom_fields[DiscourseTranslator::DETECTED_LANG_CUSTOM_FIELD]).not_to be_nil
+    end
+
+    expect(Discourse.redis.smembers(DiscourseTranslator::LANG_DETECT_NEEDED)).to be_empty
+  end
+
+  it "does not process posts if the translator is disabled" do
+    SiteSetting.translator_enabled = false
+    described_class.new.execute({})
+
+    posts.each do |post|
+      post.reload
+      expect(post.custom_fields[DiscourseTranslator::DETECTED_LANG_CUSTOM_FIELD]).to be_nil
+    end
+
+    expect(Discourse.redis.smembers(DiscourseTranslator::LANG_DETECT_NEEDED)).to match_array(posts.map(&:id).map(&:to_s))
+  end
+
+  it "processes a maximum of MAX_QUEUE_SIZE posts per run" do
+    large_number = 2000
+    large_number.times { |i| Discourse.redis.sadd?(DiscourseTranslator::LANG_DETECT_NEEDED, i + 1) }
+    described_class.new.execute({})
+
+    remaining = Discourse.redis.scard(DiscourseTranslator::LANG_DETECT_NEEDED)
+    expect(remaining).to eq(large_number - Jobs::DetectPostsTranslation::MAX_QUEUE_SIZE)
+  end
+
+  it "handles an empty Redis queue gracefully" do
+    Discourse.redis.del(DiscourseTranslator::LANG_DETECT_NEEDED)
+    expect { described_class.new.execute({}) }.not_to raise_error
+  end
+
+  it "removes successfully processed posts from Redis" do
+    described_class.new.execute({})
+
+    posts.each do |post|
+      expect(Discourse.redis.sismember(DiscourseTranslator::LANG_DETECT_NEEDED, post.id)).to be_falsey
+    end
+  end
+
+  it "skips posts that no longer exist" do
+    non_existent_post_id = -1
+    Discourse.redis.sadd?(DiscourseTranslator::LANG_DETECT_NEEDED, non_existent_post_id)
+
+    expect { described_class.new.execute({}) }.not_to raise_error
+
+    expect(Discourse.redis.sismember(DiscourseTranslator::LANG_DETECT_NEEDED, non_existent_post_id)).to be_falsey
+  end
+
+  it "ensures posts are processed within a distributed mutex" do
+    mutex_spy = instance_spy(DistributedMutex)
+    allow(DistributedMutex).to receive(:synchronize).and_yield
+
+    described_class.new.execute({})
+
+    posts.each do |post|
+      expect(DistributedMutex).to have_received(:synchronize).with("detect_translation_#{post.id}")
+    end
+  end
+end

spec/serializers/post_serializer_spec.rb

@@ -77,10 +77,13 @@
               expect(serializer.can_translate).to eq(false)
             end
 
-            it "enqueues detect translation job" do
-              expect { serializer.can_translate }.to change {
-                Jobs::DetectTranslation.jobs.size
-              }.by(1)
+            it "adds post id to redis if detected_language is blank" do
+              post.custom_fields["detected_language"] = nil
+              post.save_custom_fields
+
+              serializer.can_translate
+
+              expect(Discourse.redis.sismember(DiscourseTranslator::LANG_DETECT_NEEDED, post.id)).to eq(true)
             end
           end