ActiveRecord Row Lock Test

row_lock_test.rb

require 'rubygems'
require 'active_record'
require 'logger'

# Run as `ruby row_lock_test.rb [mysql|postgresql]` to see current Rails behavior when two processes
# try to destroy a record at the same time.
#
# Run as `ruby row_lock_test.rb [mysql|postgresql] patch` to see the behavior desired in pull request
# https://github.com/rails/rails/pull/7965
#
# You may need to edit the connection info below to match your database setup.
#
# The code spawns three threads. The first two will each select and destroy the same record. The third
# thread will repeatedly select rows from the database include the record targeted for deletion. There
# are both before and after destroy callbacks on the model being deleted that write out to STDERR when
# they are executed. Each callback includes a 5 second sleep interval to force concurrency.
#
# What you should see in the current Rails implementation is Thread 1 and Thread 2 both open transactions and
# execute the before_destroy callback. Then Thread 1 should delete the record from the database and execute the
# after_destroy callback. Thread 2 should do nothing in this interval because it will wait for the
# database lock on the row it is trying to delete. After Thread 1 commits the transaction, Thread 2
# will resume and run the after_destroy callbacks. In the mean time, the third thread should constantly
# be able to select records and not hang because it is not asking for a lock on them.
#
# What you should see with the patch is Thread 1 and Thread 1 open transactions, Thread 1 obtains a lock on the
# row to be deleted and executes the before_destroy callback, deletes the row, and the after_destroy callback.
# Thread 2 should not be active during this interval as it waits for the database lock to be released on the row.
# Thread 2 should then select the row for update, find it is not there and not run the callbacks.

connection_info = {
  "mysql" => {
    "adapter" => "mysql2",
    "database" => "test",
    "username" => "root",
    "password" => nil,
    "host" => "localhost"
  },
  "postgresql" => {
    "adapter" => "postgresql",
    "database" => "test",
    "username" => "postgres",
    "password" => "",
    "host" => "localhost"
  }
}

if ARGV[1] == "patch"
  module ActiveRecord::Callbacks
    def destroy #:nodoc:
      # Select the row to delete using a database lock before running the callbacks so that callbacks
      # won't be run if the row has already been deleted by another process.
      klass = self.class
      sql = klass.where(klass.primary_key => self.id).select(klass.primary_key).lock(true).to_sql
      row_exists = !!klass.connection.select_one(sql)
      STDERR.write "Thread #{Thread.current[:thread_id]} SELECT FOR UPDATE #{row_exists ? 'did' : 'does not'} find row\n"
      if row_exists
        run_callbacks(:destroy) { super }
      else
        super
      end
    end
  end
end

connection_params = connection_info[ARGV[0]] || raise("No connection_info defined for #{ARGV[0].inspect}")
ActiveRecord::Base.establish_connection(connection_params.merge("pool" => 4))

class MyModel < ActiveRecord::Base
  connection.create_table(table_name) do |t|
    t.string :name
  end unless table_exists?
  
  # Report that the callback is being run and include a sleep to induce concurrency.
  before_destroy do
    5.times do
      STDERR.write "Thread #{Thread.current[:thread_id]} running before_destroy callback\n"
      sleep(1)
    end
  end
  
  # Report that the callback is being run and include a sleep to induce concurrency.
  after_destroy do
    5.times do
      STDERR.write "Thread #{Thread.current[:thread_id]} running after_destroy callback\n"
      sleep(1)
    end
  end
end

# Clean up the database so we have three known records in it.
MyModel.delete_all
MyModel.create!(:name => "shared record")
2.times{|i| MyModel.create!(:name => "record #{i + 1}")}

ActiveRecord::Base.logger = Logger.new(STDERR)
ActiveRecord::Base.logger.formatter = lambda do |severity, datetime, progname, msg|
  "Thread #{Thread.current[:thread_id]} #{msg}\n" if Thread.current[:thread_id]
end

threads = []
2.times do |i|
  threads << Thread.new do
    record = MyModel.find_by_name("shared record")
    Thread.current[:thread_id] = i + 1
    sleep(0.25 + (i / 10.0)) # Sleep to try to get thread 1 to go first
    record.destroy
    ActiveRecord::Base.connection.close
  end
end

selector_thread = Thread.new do
  loop do
    record = MyModel.find_by_name("shared record")
    record_1 = MyModel.find_by_name("shared record")
    record_2 = MyModel.find_by_name("shared record")
    if record
      STDERR.write("Selector Thread reports shared record exists\n")
    else
      STDERR.write("Selector Thread reports shared record deleted\n")
    end
    sleep(1)
  end
end

threads.each{|thread| thread.join}
sleep(1.1) # Sleep to give the selector thread a chance to report that the record was indeed deleted.
selector_thread.kill