Merge branch 'master' of https://github.com/CartoDB/cartodb into update-industries

NEWS.md

@@ -8,6 +8,7 @@ Development
 - None yet
 
 ### Bug fixes / enhancements
+- Improve concurrent Ghost Tables syncs handling ([#15272](https://github.com/CartoDB/cartodb/pull/15272))
 - Fix consent screen in OAuth apps without user ([#15247](https://github.com/CartoDB/cartodb/pull/15247))
 - Update user industries options with the allowed values from Hubspot ([#15265](https://github.com/CartoDB/cartodb/pull/15265))
 

lib/carto/bolt.rb

@@ -11,24 +11,35 @@ module Carto
       @ttl_ms = ttl_ms
     end
 
+    # Run a block of code with the lock acquired.
+    # It will retry acquiring the lock up to `attempts` times and
+    # for up to `timeout` milliseconds.
+    # If an executable (lambda/Proc) object is passed through `fail_function`
+    # it will be executed if the lock is not acquired and another such proc hasn't
+    # been executed during the lock period (before a new locked execution).
+    # This can be used to reschedule execution while avoiding to reschedule
+    # additional executions while one is pending.
     def run_locked(attempts: DEFAULT_RETRY_ATTEMPTS,
                    timeout: DEFAULT_RETRY_TIMEOUT,
-                   rerun_func: nil)
+                   fail_function: nil)
       raise 'no code block given' unless block_given?
-      raise 'no proc/lambda passed as rerun_func' if rerun_func.present? && !proc?(rerun_func)
+      raise 'no proc/lambda passed as fail_function' if fail_function.present? && !proc?(fail_function)
 
-      locked_acquired = acquire_lock(attempts, timeout)
+      lock_acquired = acquire_lock(attempts, timeout)
 
       begin
-        unless locked_acquired
-          set_rerun_after_finish
-          return !!locked_acquired
+        if lock_acquired
+          retried
+          yield
+          true
+        else
+          if fail_function && !set_retry
+            fail_function.call
+          end
+          false
         end
-        yield
-        try_to_rerun(rerun_func)
-        !!locked_acquired
       ensure
-        unlock if locked_acquired
+        unlock if lock_acquired
       end
     end
 
@@ -48,14 +59,6 @@ module Carto
       false
     end
 
-    def try_to_rerun(rerun_func)
-      return unless rerun_func.present?
-      while retry?
-        refresh_lock_timeout
-        rerun_func.call
-      end
-    end
-
     def proc?(proc)
       proc.respond_to?(:call)
     end
@@ -75,16 +78,12 @@ module Carto
       @redis_object.set(@bolt_key, true, px: @ttl_ms, nx: true)
     end
 
-    def set_rerun_after_finish
-      @redis_object.set("#{@bolt_key}:retry", true, px: @ttl_ms, nx: true)
-    end
-
-    def refresh_lock_timeout
-      @redis_object.pexpire(@bolt_key, @ttl_ms)
+    def retried
+      @redis_object.del("#{@bolt_key}:retry")
     end
 
-    def retry?
-      @redis_object.del("#{@bolt_key}:retry") > 0
+    def set_retry
+      @redis_object.getset("#{@bolt_key}:retry", true)
     end
 
     def add_namespace_to_key(key)

lib/carto/ghost_tables_manager.rb

@@ -20,7 +20,7 @@ module Carto
       if should_run_synchronously?
         link_ghost_tables_synchronously
       else
-        ::Resque.enqueue(::Resque::UserDBJobs::UserDBMaintenance::LinkGhostTables, @user_id)
+        link_ghost_tables_asynchronously
       end
     end
 
@@ -28,6 +28,10 @@ module Carto
       sync_user_tables_with_db unless user_tables_synced_with_db?
     end
 
+    def link_ghost_tables_asynchronously
+      ::Resque.enqueue(::Resque::UserDBJobs::UserDBMaintenance::LinkGhostTables, @user_id)
+    end
+
     # determine linked tables vs cartodbfied tables consistency; i.e.: needs to run
     def user_tables_synced_with_db?
       user_tables = fetch_user_tables
@@ -49,8 +53,7 @@ module Carto
     def self.run_synchronized(user_id, attempts: 10, timeout: 30000, **warning_params)
       gtm = new(user_id)
       bolt = gtm.get_bolt
-      rerun_func = lambda { gtm.send(:sync) }
-      lock_acquired = bolt.run_locked(attempts: attempts, timeout: timeout, rerun_func: rerun_func) do
+      lock_acquired = bolt.run_locked(attempts: attempts, timeout: timeout) do
         yield
       end
       if !lock_acquired && warning_params.present?
@@ -74,7 +77,7 @@ module Carto
     end
 
     def sync_user_tables_with_db
-      got_locked = get_bolt.run_locked(rerun_func: lambda { sync }) { sync }
+      got_locked = get_bolt.run_locked(fail_function: lambda { link_ghost_tables_asynchronously }) { sync }
     end
 
     def sync

spec/lib/carto/bolt_spec.rb

@@ -7,6 +7,10 @@ module Carto
       @bolt = Carto::Bolt.new('manolo_bolt_locked')
     end
 
+    after(:each) do
+      @bolt.send :retried
+    end
+
     it 'should expect block' do
       expect { @bolt.run_locked.should_raise }.to raise_error('no code block given')
     end
@@ -57,50 +61,50 @@ module Carto
     end
 
     it 'should retry an execution when other process tries to acquire bolt and has retriable flag set' do
+      flag = 0
+      rerun_func = lambda { flag += 1 }
       main = Thread.new do
-        flag = 0
-        @bolt.run_locked(rerun_func: lambda { flag += 1 }) {
+        @bolt.run_locked(fail_function: rerun_func) {
           flag += 1
           sleep(2)
         }.should be_true
-        flag.should eq(2)
       end
       sleep(0.5)
       thr = Thread.new do
-        Carto::Bolt.new('manolo_bolt_locked').run_locked {}.should be_false
+        Carto::Bolt.new('manolo_bolt_locked').run_locked(fail_function: rerun_func) {}.should be_false
       end
       thr.join
       main.join
+      flag.should eq(2)
     end
 
-    it 'should execute once the rerun_func part despite of the number of calls to acquire the lock' do
+    it 'should execute once the fail_function part despite of the number of calls to acquire the lock' do
+      flag = 0
+      rerun_func = lambda do
+        flag += 1
+      end
       main = Thread.new do
-        flag = 0
-        rerun_func = lambda do
-          sleep(1)
-          flag += 1
-        end
-        @bolt.run_locked(rerun_func: rerun_func) {
+        @bolt.run_locked(fail_function: rerun_func) {
           flag += 1
           sleep(2)
         }.should be_true
-        flag.should > 2
       end
       sleep(0.5)
-      20.times do
+      10.times do
         t = Thread.new do
-          Carto::Bolt.new('manolo_bolt_locked').run_locked {}
-          sleep(0.25)
+          Carto::Bolt.new('manolo_bolt_locked').run_locked(fail_function: rerun_func) {}
+          sleep(0.1)
         end
         t.join
       end
       main.join
+      flag.should eq(2)
     end
 
-    it 'should raise error if rerun_func is not a lambda' do
+    it 'should raise error if fail_function is not a lambda' do
       expect {
-        @bolt.run_locked(rerun_func: "lala") {}.should_raise
-      }.to raise_error('no proc/lambda passed as rerun_func')
+        @bolt.run_locked(fail_function: "lala") {}.should_raise
+      }.to raise_error('no proc/lambda passed as fail_function')
     end
 
     it 'should expire a lock after ttl_ms' do

spec/lib/carto/ghost_tables_manager_spec.rb

@@ -360,7 +360,7 @@ module Carto
       ::Resque::UserDBJobs::UserDBMaintenance::LinkGhostTablesByUsername.perform(@user.username)
     end
 
-    it 'should call the rerun_func and execute sync twice becuase other worker tried to get the lock' do
+    it 'should call the fail_function and execute sync twice because other worker tried to get the lock' do
       @user.tables.count.should eq 0
       @ghost_tables_manager.instance_eval { user_tables_synced_with_db? }.should be_true
       main = Thread.new do
@@ -372,7 +372,7 @@ module Carto
           Carto::GhostTablesManager.new(@user.id).send(:sync)
         end
         gtm = Carto::GhostTablesManager.new(@user.id)
-        gtm.get_bolt.run_locked(rerun_func: rerun_func) do
+        gtm.get_bolt.run_locked(fail_function: rerun_func) do
           sleep(1)
           Carto::GhostTablesManager.new(@user.id).send(:sync)
         end

spec/models/user_organization_spec.rb

@@ -83,7 +83,9 @@ describe UserOrganization do
       @owner.reload
 
       triggers_after = @owner.db_service.triggers
-      triggers_after.map { |t| [t.database_name, t.table_name, t.trigger_name] } .should == triggers_before.map { |t| [t.database_name, t.table_name, t.trigger_name] }
+      triggers_after.map { |t| [t.database_name, t.table_name, t.trigger_name] } \
+        .should =~ \
+      triggers_before.map { |t| [t.database_name, t.table_name, t.trigger_name] }
 
       @owner.db_service.triggers('public').should be_empty
     end