Fix deadlock when pushing multiple gem versions concurrently

app/jobs/after_version_write_job.rb

@@ -9,7 +9,6 @@ def perform(version:)
       version.rubygem.push_notifiable_owners.each do |notified_user|
         Mailer.gem_pushed(owner, version.id, notified_user.id).deliver_later
       end
-      Indexer.perform_later
       UploadVersionsFileJob.perform_later
       UploadInfoFileJob.perform_later(rubygem_name: rubygem.name)
       UploadNamesFileJob.perform_later
@@ -21,7 +20,7 @@ def perform(version:)
       version.info_checksum_v2 = gem_info.info_checksum
       version.save(validate: false)
 
-      SetLinksetHomeJob.perform_later(version:)
+      ReorderVersionsJob.perform_later(rubygem:)
     end
   end
 

app/jobs/reorder_versions_job.rb

@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+class ReorderVersionsJob < ApplicationJob
+  include GoodJob::ActiveJobExtensions::Concurrency
+
+  good_job_control_concurrency_with(
+    perform_limit: 1,
+    key: -> { "reorder-versions-#{arguments.first[:rubygem].id}" }
+  )
+
+  queue_as :default
+
+  retry_on ActiveRecord::Deadlocked, wait: :polynomially_longer, attempts: 3
+  discard_on ActiveJob::DeserializationError
+
+  def perform(rubygem:)
+    logger.info { "Reordering versions for gem: #{rubygem.name} (#{rubygem.id})" }
+
+    StatsD.measure("reorder_versions.duration") do
+      rubygem.reorder_versions
+    end
+    StatsD.increment("reorder_versions.success")
+    Indexer.perform_later
+
+    latest_version = rubygem.reload.most_recent_version
+    SetLinksetHomeJob.perform_later(version: latest_version) if latest_version
+
+    logger.info { "Reordering complete for #{rubygem.name}" }
+  rescue StandardError => e
+    logger.error { "Failed to reorder versions for #{rubygem.name}: #{e.message}" }
+    StatsD.increment("reorder_versions.error", tags: { error: e.class.name })
+    raise
+  end
+end

app/models/deletion.rb

@@ -102,7 +102,7 @@ def restore_to_index
   end
 
   def reindex
-    Indexer.perform_later
+    ReorderVersionsJob.perform_later(rubygem: version.rubygem)
     UploadInfoFileJob.perform_later(rubygem_name: rubygem_name)
     UploadVersionsFileJob.perform_later
     UploadNamesFileJob.perform_later

app/models/rubygem.rb

@@ -20,7 +20,7 @@ class Rubygem < ApplicationRecord
   has_many :audits, as: :auditable, inverse_of: :auditable
   has_many :link_verifications, as: :linkable, inverse_of: :linkable, dependent: :destroy
   has_many :oidc_rubygem_trusted_publishers, class_name: "OIDC::RubygemTrustedPublisher", inverse_of: :rubygem, dependent: :destroy
-  has_many :incoming_dependencies, -> { where(versions: { indexed: true, position: 0 }) }, class_name: "Dependency", inverse_of: :rubygem
+  has_many :incoming_dependencies, -> { where(versions: { indexed: true, latest: true }) }, class_name: "Dependency", inverse_of: :rubygem
   has_many :reverse_dependencies, through: :incoming_dependencies, source: :version_rubygem
   has_many :reverse_development_dependencies, -> { merge(Dependency.development) }, through: :incoming_dependencies, source: :version_rubygem
   has_many :reverse_runtime_dependencies, -> { merge(Dependency.runtime) }, through: :incoming_dependencies, source: :version_rubygem
@@ -29,7 +29,7 @@ def unique_reverse_dependencies
     Rubygem.where(
       id: Dependency.where(rubygem_id: id)
         .joins(:version)
-        .where(versions: { indexed: true, position: 0 })
+        .where(versions: { indexed: true, latest: true })
         .select("versions.rubygem_id")
     )
   end
@@ -38,7 +38,7 @@ def unique_reverse_development_dependencies
     Rubygem.where(
       id: Dependency.development.where(rubygem_id: id)
         .joins(:version)
-        .where(versions: { indexed: true, position: 0 })
+        .where(versions: { indexed: true, latest: true })
         .select("versions.rubygem_id")
     )
   end
@@ -47,7 +47,7 @@ def unique_reverse_runtime_dependencies
     Rubygem.where(
       id: Dependency.runtime.where(rubygem_id: id)
         .joins(:version)
-        .where(versions: { indexed: true, position: 0 })
+        .where(versions: { indexed: true, latest: true })
         .select("versions.rubygem_id")
     )
   end
@@ -61,9 +61,12 @@ def unique_reverse_runtime_dependencies
   has_one :most_recent_version,
     lambda {
       order(
+        # During the async reorder window, a freshly pushed version can have a nil
+        # position; treat it as newest until ReorderVersionsJob assigns positions.
+        Arel.sql("case when #{quoted_table_name}.indexed then 0 else 1 end"),
         Arel.sql("case when #{quoted_table_name}.latest AND #{quoted_table_name}.platform = 'ruby' then 0 " \
                  "when #{quoted_table_name}.latest then 1 else 2 end"),
-        :position,
+        Arel.sql("#{quoted_table_name}.position ASC NULLS FIRST"),
         id: :desc
       )
     },
@@ -433,7 +436,7 @@ def bulk_reorder_versions
 
     ids = []
     positions = []
-    versions.each do |version|
+    versions.order(:id).each do |version|
       ids << version.id
       positions << numbers.index(version.number)
     end

app/models/version.rb

@@ -27,7 +27,6 @@ class Version < ApplicationRecord # rubocop:disable Metrics/ClassLength
   # TODO: Remove this once we move to GemDownload only
   after_create :create_gem_download
   after_create :record_push_event
-  after_save :reorder_versions, if: -> { saved_change_to_indexed? || saved_change_to_id? }
   after_save :enqueue_web_hook_jobs, if: -> { saved_change_to_indexed? && (!saved_change_to_id? || indexed?) }
   after_save :refresh_rubygem_indexed, if: -> { saved_change_to_indexed? || saved_change_to_id? }
 
@@ -255,10 +254,12 @@ def refresh_rubygem_indexed
   end
 
   def previous
+    return nil if position.nil?
     rubygem.versions.find_by(position: position + 1)
   end
 
   def next
+    return nil if position.nil?
     rubygem.versions.find_by(position: position - 1)
   end
 

test/factories/version.rb

@@ -41,6 +41,8 @@
         checksum = GemInfo.new(version.rubygem.name).info_checksum
         version.update_attribute :info_checksum_v2, checksum
       end
+
+      version.rubygem.reorder_versions
     end
   end
 end

test/functional/api/v1/deletions_controller_test.rb

@@ -374,7 +374,7 @@ class Api::V1::DeletionsControllerTest < ActionController::TestCase
           assert_enqueued_jobs 1, only: NotifyWebHookJob
         end
         should "have enqueued reindexing job" do
-          assert_enqueued_jobs 1, only: Indexer
+          assert_enqueued_jobs 1, only: ReorderVersionsJob
           assert_enqueued_jobs 1, only: UploadVersionsFileJob
           assert_enqueued_jobs 1, only: UploadNamesFileJob
           assert_enqueued_with job: UploadInfoFileJob, args: [rubygem_name: @rubygem.name]

test/functional/api/v1/rubygems_controller_test.rb

@@ -356,7 +356,7 @@ def self.should_respond_to(format)
           assert_enqueued_jobs 1, only: ActionMailer::MailDeliveryJob do
             assert_enqueued_jobs 6, only: FastlyPurgeJob do
               assert_enqueued_jobs 1, only: NotifyWebHookJob do
-                assert_enqueued_jobs 1, only: Indexer do
+                assert_enqueued_jobs 1, only: ReorderVersionsJob do
                   assert_enqueued_jobs 1, only: ReindexRubygemJob do
                     post :create, body: gem_file("test-1.0.0.gem", &:read)
                   end
@@ -572,7 +572,9 @@ def self.should_respond_to(format)
         setup do
           @user.enable_totp!(ROTP::Base32.random_base32, :ui_and_api)
           @request.env["HTTP_OTP"] = ROTP::TOTP.new(@user.totp_seed).now
-          post :create, body: gem_file("test-1.0.0.gem", &:read)
+          perform_enqueued_jobs(only: ReorderVersionsJob) do
+            post :create, body: gem_file("test-1.0.0.gem", &:read)
+          end
         end
 
         should respond_with :success
@@ -581,7 +583,7 @@ def self.should_respond_to(format)
           assert_equal 2, Rubygem.last.versions.count
         end
         should "disable mfa requirement" do
-          refute_predicate @rubygem, :metadata_mfa_required?
+          refute_predicate @rubygem.reload, :metadata_mfa_required?
         end
       end
     end

test/integration/push_test.rb

@@ -82,7 +82,8 @@ class PushTest < ActionDispatch::IntegrationTest
     push_gem gem_io
 
     assert_response :success
-    perform_enqueued_jobs
+    perform_enqueued_jobs(only: ReorderVersionsJob)
+    perform_enqueued_jobs(only: SetLinksetHomeJob)
 
     get rubygem_path("sandworm")
 

test/integration/pusher_test.rb

@@ -423,7 +423,7 @@ def two_cert_chain(signing_key:, root_not_before: Time.current, cert_not_before:
     should "enqueue job for email, updating ES index, spec index and purging cdn" do
       assert_enqueued_jobs 1, only: ActionMailer::MailDeliveryJob do
         assert_enqueued_jobs 6, only: FastlyPurgeJob do
-          assert_enqueued_jobs 1, only: Indexer do
+          assert_enqueued_jobs 1, only: ReorderVersionsJob do
             assert_enqueued_jobs 1, only: ReindexRubygemJob do
               @cutter.save
             end

test/jobs/reorder_versions_job_test.rb

@@ -0,0 +1,85 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class ReorderVersionsJobTest < ActiveJob::TestCase
+  setup do
+    @rubygem = create(:rubygem, name: "test-gem")
+    @user = create(:user)
+  end
+
+  context "reordering versions" do
+    should "reorder versions, set the latest flag, and record the success metric" do
+      v3 = create(:version, rubygem: @rubygem, number: "3.0.0", indexed: true)
+      v1 = create(:version, rubygem: @rubygem, number: "1.0.0", indexed: true)
+      v2 = create(:version, rubygem: @rubygem, number: "2.0.0", indexed: true)
+
+      StatsD.stubs(:increment)
+      StatsD.stubs(:measure)
+      StatsD.expects(:increment).with("reorder_versions.success")
+      StatsD.expects(:measure).with("reorder_versions.duration").yields
+
+      assert_enqueued_with(job: Indexer) do
+        ReorderVersionsJob.new.perform(rubygem: @rubygem)
+      end
+
+      assert_equal 0, v3.reload.position
+      assert_equal 1, v2.reload.position
+      assert_equal 2, v1.reload.position
+
+      refute v1.reload.latest
+      refute v2.reload.latest
+      assert v3.reload.latest
+    end
+
+    should "handle concurrent reorder attempts gracefully" do
+      create(:version, rubygem: @rubygem, number: "1.0.0", indexed: true)
+
+      job1 = ReorderVersionsJob.new
+      job2 = ReorderVersionsJob.new
+
+      assert_nothing_raised do
+        threads = [
+          Thread.new do
+            ActiveRecord::Base.connection_pool.with_connection do
+              job1.perform(rubygem: @rubygem)
+            end
+          end,
+          Thread.new do
+            ActiveRecord::Base.connection_pool.with_connection do
+              job2.perform(rubygem: @rubygem)
+            end
+          end
+        ]
+        threads.each(&:join)
+      end
+
+      assert_equal 0, @rubygem.versions.first.reload.position
+    end
+
+    should "handle errors and increment error metric" do
+      create(:version, rubygem: @rubygem, number: "1.0.0", indexed: true)
+
+      @rubygem.stubs(:reorder_versions).raises(StandardError.new("Test error"))
+
+      StatsD.stubs(:increment)
+      StatsD.stubs(:measure).yields
+      StatsD.expects(:increment).with("reorder_versions.error", tags: { error: "StandardError" })
+
+      assert_raises(StandardError) do
+        ReorderVersionsJob.new.perform(rubygem: @rubygem)
+      end
+    end
+
+    should "discard job if rubygem no longer exists" do
+      rubygem_id = @rubygem.id
+      @rubygem.destroy
+
+      assert_nothing_raised do
+        ReorderVersionsJob.perform_now(rubygem: Rubygem.find(rubygem_id))
+      rescue ActiveRecord::RecordNotFound, ActiveJob::DeserializationError => e
+        Rails.logger.info "Job discarded as expected: #{e.class}"
+      end
+    end
+  end
+end

test/models/deletion_test.rb

@@ -162,7 +162,7 @@ class DeletionTest < ActiveSupport::TestCase
   should "enque job for updating ES index, spec index and purging cdn" do
     assert_enqueued_jobs 1, only: ActionMailer::MailDeliveryJob do
       assert_enqueued_jobs 8, only: FastlyPurgeJob do
-        assert_enqueued_jobs 1, only: Indexer do
+        assert_enqueued_jobs 1, only: ReorderVersionsJob do
           assert_enqueued_jobs 1, only: ReindexRubygemJob do
             delete_gem
           end
@@ -213,6 +213,8 @@ class DeletionTest < ActiveSupport::TestCase
       end
 
       should "reorder versions" do
+        perform_enqueued_jobs(only: ReorderVersionsJob)
+
         assert_predicate @version.reload, :latest?
       end
 
@@ -271,7 +273,7 @@ class DeletionTest < ActiveSupport::TestCase
 
     should "enqueue indexing jobs" do
       @deletion = delete_gem
-      assert_enqueued_jobs 1, only: Indexer do
+      assert_enqueued_jobs 1, only: ReorderVersionsJob do
         assert_enqueued_jobs 1, only: UploadVersionsFileJob do
           assert_enqueued_with job: UploadInfoFileJob, args: [rubygem_name: @gem_name] do
             @deletion.restore!

test/models/version_test.rb

@@ -1095,26 +1095,4 @@ class VersionTest < ActiveSupport::TestCase
       assert_does_not_contain Version.created_between(@start_time, @end_time), @version
     end
   end
-
-  context "after_save" do
-    context "reorder versions" do
-      setup do
-        @version = create(:version)
-      end
-
-      context "indexed is updated" do
-        should "reorder versions" do
-          @version.expects(:reorder_versions).times(1)
-          @version.update(indexed: false)
-        end
-      end
-
-      context "info checksum v2 is updated" do
-        should "not reorder versions" do
-          @version.expects(:reorder_versions).times(0)
-          @version.update(info_checksum_v2: "lala")
-        end
-      end
-    end
-  end
 end

test/system/avo/versions_test.rb

@@ -74,7 +74,7 @@ class Avo::VersionsSystemTest < ApplicationSystemTestCase
             },
             "unchanged" => version_attributes
               .except("updated_at", "yanked_info_checksum_v2", "yanked_at", "indexed")
-              .merge("position" => 0, "latest" => false)
+              .merge("position" => 0, "latest" => true)
               .transform_values(&:as_json)
           },
           "gid://gemcutter/Rubygem/#{rubygem.id}" =>

test/system/gem_server_lifecycle_test.rb

@@ -43,6 +43,7 @@ class GemServerLifecycleTest < ApplicationSystemTestCase
 
     Indexer.perform_now
     @subscriber = ActiveSupport::Notifications.subscribe("process_action.action_controller") do
+      perform_enqueued_jobs only: [ReorderVersionsJob]
       perform_enqueued_jobs only: [Indexer]
     end
 

test/test_helper.rb

@@ -80,6 +80,7 @@
 WebAuthn.configuration.allowed_origins = ["http://localhost:31337"]
 
 class ActiveSupport::TestCase
+  include ActiveJob::TestHelper
   include FactoryBot::Syntax::Methods
   include GemHelpers
   include EmailHelpers