Add new option. Review.

Author: omeH

lib/table_sync/receiving/hooks/once.rb

@@ -2,6 +2,8 @@
 
 module TableSync::Receiving::Hooks
   class Once
+    LOCK_KEY = "hook-once-lock-key"
+
     attr_reader :conditions, :handler, :lookup_code
 
     def initialize(conditions:, handler:)
@@ -21,8 +23,9 @@ def perform(config:, targets:)
       targets.each do |target|
         next unless conditions?(target)
 
-        model.transaction(isolation: model.isolation_level(:repeatable)) do
-          model.find_and_save(row: target, target_keys:) do |entry|
+        keys = target.slice(*target_keys)
+        model.try_advisory_lock(prepare_lock_key(keys)) do
+          model.find_and_save(keys:) do |entry|
             next unless allow?(entry)
 
             entry.hooks ||= []
@@ -50,5 +53,10 @@ def conditions?(row)
         row[column] == (conditions[column] || row[column])
       end
     end
+
+    def prepare_lock_key(row_keys)
+      lock_keys = [LOCK_KEY] + row_keys.values
+      Zlib.crc32(lock_keys.join(":")) % (2**31)
+    end
   end
 end

lib/table_sync/receiving/model/active_record.rb

@@ -129,8 +129,16 @@ def after_commit(&)
       db.add_transaction_record(AfterCommitWrap.new(&))
     end
 
-    def find_and_save(row:, target_keys:)
-      entry = raw_model.find_by(row.slice(*target_keys))
+    def try_advisory_lock(lock_key)
+      transaction do
+        if db.query_value("SELECT pg_try_advisory_xact_lock(#{lock_key.to_i})")
+          yield
+        end
+      end
+    end
+
+    def find_and_save(keys:)
+      entry = raw_model.find_by(keys)
       return unless entry
 
       yield entry

lib/table_sync/receiving/model/sequel.rb

@@ -76,8 +76,16 @@ def after_commit(&)
       db.after_commit(&)
     end
 
-    def find_and_save(row:, target_keys:)
-      entry = dataset.first(row.slice(*target_keys))
+    def try_advisory_lock(lock_key)
+      transaction do
+        if db.get(::Sequel.function(:pg_try_advisory_xact_lock, lock_key.to_i))
+          yield
+        end
+      end
+    end
+
+    def find_and_save(keys:)
+      entry = dataset.first(keys)
       return unless entry
 
       yield entry

spec/receiving/models/active_record_spec.rb

@@ -23,18 +23,18 @@
       )
     end
 
-    let(:row) { { external_id: } }
+    let(:keys) { { external_id: } }
 
     it "finds and saves an entry" do
-      model.find_and_save(row:, target_keys: [primary_key]) do |entry|
+      model.find_and_save(keys:) do |entry|
         entry.online_status = true
       end
       expect(player.reload.online_status).to be_truthy
     end
 
     it "does nothing" do
-      row = { external_id: external_id + 1 }
-      model.find_and_save(row:, target_keys: [primary_key]) do |entry|
+      keys = { external_id: external_id + 1 }
+      model.find_and_save(keys:) do |entry|
         entry.online_status = true
       end
       expect(player.reload.online_status).to be_falsy

spec/receiving/models/sequel_spec.rb

@@ -23,18 +23,18 @@
       )
     end
 
-    let(:row) { { external_id: } }
+    let(:keys) { { external_id: } }
 
     it "finds and saves an entry" do
-      model.find_and_save(row:, target_keys: [primary_key]) do |entry|
+      model.find_and_save(keys:) do |entry|
         entry.online_status = true
       end
       expect(player.reload.online_status).to be_truthy
     end
 
     it "does nothing" do
-      row = { external_id: external_id + 1 }
-      model.find_and_save(row:, target_keys: [primary_key]) do |entry|
+      keys = { external_id: external_id + 1 }
+      model.find_and_save(keys:) do |entry|
         entry.online_status = true
       end
       expect(player.reload.online_status).to be_falsy