Make leader election idempotent

Author: ChrisBr

ruby/lib/ci/queue/redis/base.rb

@@ -8,6 +8,7 @@ class Base
         TEN_MINUTES = 60 * 10
         CONNECTION_ERRORS = [
           ::Redis::BaseConnectionError,
+          RedisClient::ConnectionError,
           ::SocketError, # https://github.com/redis/redis-rb/pull/631
         ].freeze
 

ruby/lib/ci/queue/redis/worker.rb

@@ -201,43 +201,69 @@ def try_to_reserve_lost_test
         def push(tests)
           @total = tests.size
 
-          if @master = redis.setnx(key('master-status'), 'setup')
-            puts "Worker electected as leader, pushing #{@total} tests to the queue."
-            puts
-
-            attempts = 0
-            duration = measure do
-              with_redis_timeout(5) do
-                redis.without_reconnect do
-                  redis.multi do |transaction|
-                    transaction.lpush(key('queue'), tests) unless tests.empty?
-                    transaction.set(key('total'), @total)
-                    transaction.set(key('master-status'), 'ready')
-
-                    transaction.expire(key('queue'), config.redis_ttl)
-                    transaction.expire(key('total'), config.redis_ttl)
-                    transaction.expire(key('master-status'), config.redis_ttl)
+          with_redis_timeout(5) do
+            redis.without_reconnect do
+              @master = leader_election do
+                puts "Worker electected as leader, pushing #{@total} tests to the queue."
+                puts
+
+                attempts = 0
+                duration = measure do
+                  begin
+                    redis.multi do |transaction|
+                      transaction.lpush(key('queue'), tests) unless tests.empty?
+                      transaction.set(key('total'), @total)
+                      transaction.set(key('master-status'), 'ready')
+
+                      transaction.expire(key('queue'), config.redis_ttl)
+                      transaction.expire(key('total'), config.redis_ttl)
+                      transaction.expire(key('master-status'), config.redis_ttl)
+                    end
+                  rescue ::Redis::BaseError, RedisClient::Error => error
+                    if !queue_initialized? && attempts < 3
+                      puts "Retrying pushing #{@total} tests to the queue... (#{error})"
+                      attempts += 1
+                      retry
+                    end
+
+                    raise if !queue_initialized?
                   end
                 end
-              rescue ::Redis::BaseError => error
-                if !queue_initialized? && attempts < 3
-                  puts "Retrying pushing #{@total} tests to the queue... (#{error})"
-                  attempts += 1
-                  retry
-                end
-
-                raise if !queue_initialized?
+                puts "Finished pushing #{@total} tests to the queue in #{duration.round(2)}s."
               end
             end
-
-            puts "Finished pushing #{@total} tests to the queue in #{duration.round(2)}s."
           end
           register
           redis.expire(key('workers'), config.redis_ttl)
         rescue *CONNECTION_ERRORS
           raise if @master
         end
 
+        def leader_election
+          attempts = 0
+          value = key('setup')
+
+          begin
+            if master = redis.setnx(key('master-status'), value)
+              yield
+            end
+          rescue ::Redis::BaseError, RedisClient::Error => error
+            puts "Error during leader election: #{error}"
+            if redis.get(key('master-status')) == value
+              master = true
+              yield
+            elsif attempts < 3
+              puts "Retrying leader election... (#{error})"
+              attempts += 1
+              retry
+            else
+              raise error
+            end
+          end
+
+          master
+        end
+
         def register
           redis.sadd(key('workers'), [worker_id])
         end