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

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,43 @@ 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) { |batch| process_batch(batch) }
+      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 +198,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,90 @@
+# 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 { |post| Discourse.redis.sadd?(DiscourseTranslator::LANG_DETECT_NEEDED, post.id) }
+  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,15 @@
               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