#424

backend/spec/services/shared_resource_id_allocator_mysql_spec.rb

@@ -0,0 +1,143 @@
+require 'rails_helper'
+require 'timeout'
+
+RSpec.describe 'Shared resource IDs across MySQL connections', type: :model do
+  self.use_transactional_tests = false
+
+  before do
+    expect(Rails.env.test?).to be(true)
+    expect(ApplicationRecord.connection.adapter_name).to match(/mysql/i)
+    expect(ApplicationRecord.connection_pool.size).to be >= 3
+    @prefix = "shared_ids_#{ SecureRandom.hex(8) }"
+    @names = [0, 1].map { create(:tag_name, name: "#{ @prefix }_#{ _1 }") }
+    @tag = Tag.create!(tag_name_id: @names.first.id, category: :general)
+    @names.first.update!(tag_id: @tag.id)
+    @threads = []
+    @release = Queue.new
+  end
+
+  after do
+    @release << true if @release
+    begin
+      Array(@threads).each do |thread|
+        raise 'Shared ID worker did not finish' unless thread.join(15)
+      end
+    ensure
+      if @prefix
+        ExternalTag.where(name: "#{ @prefix }_external").delete_all
+        TagGroup.where(name: "#{ @prefix }_group").delete_all
+      end
+      TagDescription.where(tag_id: @tag.id).delete_all if @tag
+      if @names
+        ids = @names.map(&:id)
+        TagName.where(id: ids).update_all(tag_id: nil)
+        Tag.where(tag_name_id: ids).delete_all
+        TagName.where(id: ids).delete_all
+      end
+    end
+  end
+
+  def await_queue queue
+    Timeout.timeout(15) { queue.pop }
+  end
+
+  def worker &block
+    Thread.new do
+      Thread.current.report_on_exception = false
+      ApplicationRecord.connection_pool.with_connection(&block)
+    end.tap { @threads << _1 }
+  end
+
+  def create_resource kind
+    case kind
+    when :tag
+      Tag.create!(tag_name_id: @names.last.id, category: :general)
+    when :external
+      ExternalTag.create!(platform: :nico, name: "#{ @prefix }_external")
+    when :description
+      TagDescription.create!(tag: @tag, language_code: 'ja', body: '', version_no: 1)
+    when :group
+      TagGroup.create!(name: "#{ @prefix }_group", version_no: 1)
+    end
+  end
+
+  def hold_creation kind, rollback: false
+    ready = Queue.new
+    thread = worker do |connection|
+      id = nil
+      connection.transaction do
+        lock_name = nil
+        subscriber = lambda do |*args|
+          match = args.last.fetch(:sql).match(/GET_LOCK\('([^']+)'/)
+          lock_name = match[1] if match
+        end
+        ActiveSupport::Notifications.subscribed(subscriber, 'sql.active_record') do
+          id = create_resource(kind).id
+        end
+        ready << [id, connection.select_value('SELECT CONNECTION_ID()').to_i, lock_name]
+        await_queue(@release)
+        raise ActiveRecord::Rollback if rollback
+      end
+      id
+    end
+    [thread, await_queue(ready)]
+  end
+
+  def result thread
+    raise 'Shared ID worker did not finish' unless thread.join(15)
+    thread.value
+  end
+
+  { tags: [:external, :tag], wiki: [:description, :group] }.each do |space, kinds|
+    [false, true].each do |rollback|
+      it "serialises #{ space } inserts through #{ rollback ? 'rollback' : 'commit' }" do
+        tables = SharedResourceIdAllocator::SPACES.fetch(space.to_s)
+        maxima = tables.map do |table|
+          ApplicationRecord.connection.select_value("SELECT MAX(id) FROM #{ table }").to_i
+        end
+        first, (first_id, connection_id, lock_name) = hold_creation(kinds.first, rollback:)
+        expect(first_id).to eq(maxima.max + 1)
+        connection = ApplicationRecord.connection
+        quoted_lock = connection.quote(lock_name)
+        expect(connection.select_value("SELECT IS_USED_LOCK(#{ quoted_lock })").to_i)
+            .to eq(connection_id)
+
+        started = Queue.new
+        second = worker do |other|
+          started << other.select_value('SELECT CONNECTION_ID()').to_i
+          create_resource(kinds.last).id
+        end
+        expect(await_queue(started)).not_to eq(connection_id)
+        @release << true
+        expect(result(first)).to eq(first_id)
+        expect(result(second)).to eq(rollback ? first_id : first_id + 1)
+        expect(connection.select_value("SELECT IS_FREE_LOCK(#{ quoted_lock })").to_i).to eq(1)
+      end
+    end
+  end
+
+  it 'allows the other ID space to commit while the first space remains locked' do
+    first, (_id, connection_id, lock_name) = hold_creation(:external)
+    second = worker { create_resource(:group).id }
+    expect(result(second)).to be_positive
+    connection = ApplicationRecord.connection
+    quoted_lock = connection.quote(lock_name)
+    expect(connection.select_value("SELECT IS_USED_LOCK(#{ quoted_lock })").to_i)
+        .to eq(connection_id)
+    @release << true
+    result(first)
+  end
+
+  it 'raises on a real lock acquisition failure without inserting a record' do
+    first, (_id, _connection_id, lock_name) = hold_creation(:external, rollback: true)
+    stub_const('SharedResourceIdAllocator::LOCK_TIMEOUT', 0)
+    expect { create_resource(:external) }.to raise_error(ActiveRecord::LockWaitTimeout)
+    expect(ExternalTag.where(name: "#{ @prefix }_external")).not_to exist
+    @release << true
+    result(first)
+    connection = ApplicationRecord.connection
+    quoted_lock = connection.quote(lock_name)
+    expect(connection.select_value("SELECT IS_FREE_LOCK(#{ quoted_lock })").to_i)
+        .to eq(1)
+  end
+end

backend/spec/services/shared_resource_id_allocator_spec.rb

@@ -62,4 +62,27 @@ RSpec.describe SharedResourceIdAllocator do
     expect(sql.grep(/FOR UPDATE/)).to be_empty
     callbacks.fetch(:after_commit).call
   end
+
+  it 'uses the same named lock within each space and different locks between spaces' do
+    locks = [Tag, ExternalTag, TagDescription, TagGroup].map do |model|
+      allow(model).to receive(:connection).and_return(connection)
+      allow(model).to receive(:current_transaction).and_return(transaction)
+      allow(model).to receive(:uncached).and_yield
+      record = model.new(id: 50)
+      described_class.assign!(record)
+      statement = sql.grep(/GET_LOCK/).last
+      callbacks.fetch(:after_commit).call
+      statement
+    end
+    expect(locks[0]).to eq(locks[1])
+    expect(locks[2]).to eq(locks[3])
+    expect(locks[0]).not_to eq(locks[2])
+  end
+
+  it 'reports lock release failures' do
+    described_class.assign!(record)
+    allow(connection).to receive(:select_value).with(/RELEASE_LOCK/).and_return(nil)
+    expect { callbacks.fetch(:after_commit).call }
+        .to raise_error(ActiveRecord::ActiveRecordError, /lock release failed/)
+  end
 end