With_advisory_lock test with Multiple threads fails

[allanohorn]

This issue was originally posted on Stack Overflow. [Stack Overflow].
(https://stackoverflow.com/questions/56146171/ruby-with-advisory-lock-test-with-multiple-threads-fails-intermittently)

Summary of the issue is that rails tests which create multiple threads then try to call an operation which uses with_advisory_lock do not seem work properly. Things that were tried:

1. Wrap with_advisory_lock block in a Transaction -> Locking behavior as expected
2. Create a transacion within the with_advisory_lock block -> Locking behavior as expected
3. Use only transaction and NO with_advisory_lock -> Locking behavior as expected
   The only thing that doesn't seem to work as expected is just using with_advisory_lock as intended.

STACK OVERFLOW TICKET
I'm using the with_advisory_lock gem to try and ensure that a record is created only once. Here's the github url to the gem.

I have the following code, which sits in an operation class that I wrote to handle creating user subscriptions:

def create_subscription_for user
  subscription = UserSubscription.with_advisory_lock("lock_%d" % user.id) do
    UserSubscription.where({ user_id: user.id }).first_or_create
  end

  # do more stuff on that subscription
end

and the accompanying test:

threads = []
user = FactoryBot.create(:user)

rand(5..10).times do
  threads << Thread.new do
    subject.create_subscription_for(user)
  end
end

threads.each(&:join)

expect(UserSubscription.count).to eq(1)

What I expect to happen:

• The first thread to get to the block acquires the lock and creates a record.
• Any other thread that gets to the block while it's being held by another thread [waits indefinitely until the lock is released] 1 (as per docs)
• As soon as the lock is released by the first thread that created the record, another thread acquires the lock and now finds the record because it was already created by the first thread.

What actually happens:

• The first thread to get to the block acquires the lock and creates a record.
• Any other thread that gets to the block while it's being held by another thread goes and executes the code in the block anyway and as a result, when running the test, it sometimes fails with a ActiveRecord::RecordNotUnique error (I have a unique index on the table that allows for a single user_subscription with the same user_id)

What is more weird is that if I add a sleep for a few hundred milliseconds in my method just before the find_or_create method, the test never fails:

def create_subscription_for user
  subscription = UserSubscription.with_advisory_lock("lock_%d" % user.id) do
    sleep 0.2
    UserSubscription.where({ user_id: user.id }).first_or_create
  end

  # do more stuff on that subscription
end

My questions are: "Why is adding the sleep 0.2 making the tests always pass?" and "Where do I look to debug this?"

Thanks!

UPDATE: Tweaking the tests a little bit causes them to always fail:

threads = []
user = FactoryBot.create(:user)

rand(5..10).times do
  threads << Thread.new do
    sleep
    subject.create_subscription_for(user)
  end
end

until threads.all? { |t| t.status == 'sleep' }
  sleep 0.1
end

threads.each(&:wakeup)
threads.each(&:join)

expect(UserSubscription.count).to eq(1)

I have also wrapped first_or_create in a transaction, which makes the test pass and everything to work as expected:

def create_subscription_for user
  subscription = UserSubscription.with_advisory_lock("lock_%d" % user.id) do
    UserSubscription.transaction do
      UserSubscription.where({ user_id: user.id }).first_or_create
    end
  end

  # do more stuff on that subscription
end

So why is wrapping first_or_create in a transaction necessary to make things work?

[mceachen]

(original author here, but realize it's been 6 years since I wrote this, and I'm no longer the maintainer).

If you're not in a transaction, (depending on the rdbms, but certainly true with MySQL), consistency guarantees are simply not present. It's never a bad idea to be in a transaction, as autocommit (at least used to be) much slower than explicit transaction boundaries.

[muxcmux]

Thanks for jumping on this one (SO op here). Can you elaborate a bit on “consistency guarantees are not present”? Does that mean that locks work only some of the time, or would it mean that in the context of Rails, acquiring db lock is not always guaranteed?

@allanohorn Thanks for the tip on ‘create_or_find_by’ - I ended up implementing something very similar that relies on catching unique constraint exceptions before I knew about ’create_or_find_by’

[mceachen]

With MySQL, if you're not in a txn, read results may not reflect the committed state of the database, depending on what engine you're using and how you've configured it.

[ismailakbudak]

I have same problem with rspec tests. When I disable with_advisory_lock, it works as expected

[caifara]

I had some problems but I've been able to resolve them by setting rspecs use_transactional_fixtures to false (as I should have done in the first place given I use database_cleaner).

I've written two tests to be able to see that w/o locks two threads run async and with locks they get synchronized. They work as expected, both letting database_cleaner use transactions or not.

RSpec.describe :with_advisory_lock, type: :model do
  def thread_1(lock, result)
    Thread.new do
      ActiveRecord::Base.with_advisory_lock(lock) do
        result << "thread_1"
        sleep 0.01
        result << "/thread_1"
      end
    end
  end

  def thread_2(lock, result)
    Thread.new do
      ActiveRecord::Base.with_advisory_lock(lock) do
        result << "thread_2"
      end
    end
  end

  it "can run two different synchronized tasks in parallel" do
    global_result = []

    10.times do
      result = []
      [
        thread_1("lock_1", result),
        thread_2("lock_2", result)
      ].each(&:join)

      global_result << result
    end

    expect(global_result).to include(%w[thread_1 thread_2 /thread_1])
  end

  it "can synchronize tasks with the same key" do
    10.times do
      result = []
      [
        thread_1("same_lock", result),
        thread_2("same_lock", result)
      ].each(&:join)

      # check thread_1 and /thread_1 are siblings
      expect(result.index("thread_1")).to eq(result.index("/thread_1") - 1)
    end
  end
end