Make queue setup idempotent

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

@@ -185,6 +185,14 @@ def max_test_failed?
 
         attr_reader :redis, :redis_url
 
+        def with_redis_timeout(timeout)
+          prev = redis._client.timeout
+          redis._client.timeout = timeout
+          yield
+        ensure
+          redis._client.timeout = prev
+        end
+
         def measure
           starting = Process.clock_gettime(Process::CLOCK_MONOTONIC)
           yield

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

@@ -207,15 +207,28 @@ def push(tests)
             puts "Worker electected as leader, pushing #{@total} tests to the queue."
             puts
 
+            attempts = 0
             duration = measure 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
+                  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
+                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?
               end
             end