Raise error when advisory lock cannot be acquired

lib/closure_tree/finders.rb

@@ -17,7 +17,7 @@ def find_or_create_by_path(path, attributes = {})
       return found if found
 
       attrs = subpath.shift
-      _ct.with_advisory_lock do
+      _ct.with_advisory_lock! do
         # shenanigans because children.create is bound to the superclass
         # (in the case of polymorphism):
         child = self.children.where(attrs).first || begin
@@ -159,7 +159,7 @@ def find_or_create_by_path(path, attributes = {})
         attr_path = _ct.build_ancestry_attr_path(path, attributes)
         find_by_path(attr_path) || begin
           root_attrs = attr_path.shift
-          _ct.with_advisory_lock do
+          _ct.with_advisory_lock! do
             # shenanigans because find_or_create can't infer that we want the same class as this:
             # Note that roots will already be constrained to this subclass (in the case of polymorphism):
             root = roots.where(root_attrs).first || _ct.create!(self, root_attrs)

lib/closure_tree/has_closure_tree.rb

@@ -12,6 +12,7 @@ def has_closure_tree(options = {})
         :numeric_order,
         :touch,
         :with_advisory_lock,
+        :advisory_lock_timeout_seconds,
         :order_belong_to
       )
 

lib/closure_tree/hierarchy_maintenance.rb

@@ -53,7 +53,7 @@ def _ct_after_save
     end
 
     def _ct_before_destroy
-      _ct.with_advisory_lock do
+      _ct.with_advisory_lock! do
         delete_hierarchy_references
         if _ct.options[:dependent] == :nullify
           self.class.find(self.id).children.find_each { |c| c.rebuild! }
@@ -63,7 +63,7 @@ def _ct_before_destroy
     end
 
     def rebuild!(called_by_rebuild = false)
-      _ct.with_advisory_lock do
+      _ct.with_advisory_lock! do
         delete_hierarchy_references unless (defined? @was_new_record) && @was_new_record
         hierarchy_class.create!(:ancestor => self, :descendant => self, :generations => 0)
         unless root?
@@ -89,7 +89,7 @@ def rebuild!(called_by_rebuild = false)
     end
 
     def delete_hierarchy_references
-      _ct.with_advisory_lock do
+      _ct.with_advisory_lock! do
         # The crazy double-wrapped sub-subselect works around MySQL's limitation of subselects on the same table that is being mutated.
         # It shouldn't affect performance of postgresql.
         # See http://dev.mysql.com/doc/refman/5.0/en/subquery-errors.html
@@ -111,7 +111,7 @@ module ClassMethods
       # Rebuilds the hierarchy table based on the parent_id column in the database.
       # Note that the hierarchy table will be truncated.
       def rebuild!
-        _ct.with_advisory_lock do
+        _ct.with_advisory_lock! do
           cleanup!
           roots.find_each { |n| n.send(:rebuild!) } # roots just uses the parent_id column, so this is safe.
         end

lib/closure_tree/numeric_deterministic_ordering.rb

@@ -125,7 +125,7 @@ def add_sibling(sibling, add_after = true)
       # Make sure self isn't dirty, because we're going to call reload:
       save
 
-      _ct.with_advisory_lock do
+      _ct.with_advisory_lock! do
         prior_sibling_parent = sibling.parent
         reorder_from_value = if prior_sibling_parent == self.parent
           [self.order_value, sibling.order_value].compact.min

lib/closure_tree/support.rb

@@ -23,6 +23,7 @@ def initialize(model_class, options)
         :dependent => :nullify, # or :destroy or :delete_all -- see the README
         :name_column => 'name',
         :with_advisory_lock => true,
+        :advisory_lock_timeout_seconds => 5,
         :numeric_order => false
       }.merge(options)
       raise ArgumentError, "name_column can't be 'path'" if options[:name_column] == 'path'
@@ -108,9 +109,9 @@ def where_eq(column_name, value)
       end
     end
 
-    def with_advisory_lock(&block)
+    def with_advisory_lock!(&block)
       if options[:with_advisory_lock]
-        model_class.with_advisory_lock(advisory_lock_name) do
+        model_class.with_advisory_lock!(advisory_lock_name, advisory_lock_options) do
           transaction { yield }
         end
       else

lib/closure_tree/support_attributes.rb

@@ -8,6 +8,10 @@ def advisory_lock_name
       Digest::SHA1.hexdigest("ClosureTree::#{base_class.name}")[0..32]
     end
 
+    def advisory_lock_options
+      { timeout_seconds: options[:advisory_lock_timeout_seconds] }.compact
+    end
+
     def quoted_table_name
       connection.quote_table_name(table_name)
     end

spec/closure_tree/parallel_spec.rb

@@ -103,7 +103,7 @@ def run_workers(worker_class = FindOrCreateWorker)
 
   it 'creates dupe roots without advisory locks' do
     # disable with_advisory_lock:
-    allow(Tag).to receive(:with_advisory_lock) { |_lock_name, &block| block.call }
+    allow(Tag).to receive(:with_advisory_lock!) { |_lock_name, &block| block.call }
     run_workers
     # duplication from at least one iteration:
     expect(Tag.where(name: @names).size).to be > @iterations
@@ -121,6 +121,12 @@ def work
   end
 
   it 'fails to deadlock while simultaneously deleting items from the same hierarchy' do
+    allow(User).to receive(:with_advisory_lock!).and_wrap_original do |method, *args, &block|
+      options = args.extract_options!
+      options[:timeout_seconds] = nil
+      method.call(*args, options, &block)
+    end
+
     target = User.find_or_create_by_path((1..200).to_a.map { |ea| ea.to_s })
     emails = target.self_and_ancestors.to_a.map(&:email).shuffle
     Parallel.map(emails, :in_threads => max_threads) do |email|

spec/closure_tree/support_spec.rb

@@ -1,14 +1,38 @@
 require 'spec_helper'
 
 RSpec.describe ClosureTree::Support do
-  let(:sut) { Tag._ct }
+  let(:model) { Tag }
+  let(:options) { {} }
+  let(:sut) { described_class.new(model, options) }
+
   it 'passes through table names without prefix and suffix' do
     expected = 'some_random_table_name'
     expect(sut.remove_prefix_and_suffix(expected)).to eq(expected)
   end
+
   it 'extracts through table name with prefix and suffix' do
     expected = 'some_random_table_name'
     tn = ActiveRecord::Base.table_name_prefix + expected + ActiveRecord::Base.table_name_suffix
     expect(sut.remove_prefix_and_suffix(tn)).to eq(expected)
   end
-end
+
+  [
+    [true, 10, { timeout_seconds: 10 }],
+    [true, nil, {}],
+    [false, nil, {}]
+  ].each do |with_lock, timeout, expected_options|
+    context "when with_advisory_lock is #{with_lock} and timeout is #{timeout}" do
+      let(:options) { { with_advisory_lock: with_lock, advisory_lock_timeout_seconds: timeout } }
+
+      it "uses advisory lock with timeout options: #{expected_options}" do
+        if with_lock
+          expect(model).to receive(:with_advisory_lock!).with(anything, expected_options).and_yield
+        else
+          expect(model).not_to receive(:with_advisory_lock!)
+        end
+
+        expect { |b| sut.with_advisory_lock!(&b) }.to yield_control
+      end
+    end
+  end
+end