@@ -1,9 +1,7 @@
 # frozen_string_literal: true
 
 require 'ci/queue/static'
-require 'concurrent/atomic/event'
 require 'concurrent/set'
-require 'securerandom'
 
 module CI
   module Queue
@@ -19,8 +17,6 @@ class << self
 
       class Worker < Base
         DEFAULT_SLEEP_SECONDS = 0.5
-        InitializationLeaseLost = Class.new(StandardError)
-        private_constant :InitializationLeaseLost
         attr_reader :total
 
         def initialize(redis, config)
@@ -41,8 +37,8 @@ def populate(tests, random: Random.new)
           # All workers need an index of tests to resolve IDs
           @index = tests.map { |t| [t.id, t] }.to_h
           @total = tests.size
-          @ordering_random = random.dup
-          initialize_queue(tests)
+          @random = random
+          initialize_queue
 
           register_worker_presence
 
@@ -70,9 +66,9 @@ def idle?
         end
 
         def poll
-          wait_for_master(timeout: config.queue_init_timeout) do
-            initialize_queue(index.values)
-          end
+          # If the elected worker dies before publishing, its setup marker
+          # expires and a waiting worker initializes the queue instead.
+          wait_for_master(timeout: config.queue_init_timeout) { initialize_queue }
           if master?
             warn "Worker #{worker_id} is the master"
           else
@@ -270,66 +266,6 @@ def heartbeat(test_or_id = nil)
 
         attr_reader :index
 
-        def initialize_queue(tests)
-          return unless acquire_master_role?
-
-          stop_renewal = Concurrent::Event.new
-          renewal = Thread.new do
-            until stop_renewal.wait(master_lease_milliseconds / 3000.0)
-              begin
-                with_master_lease do |transaction|
-                  transaction.pexpire(key('master-lease'), master_lease_milliseconds)
-                end
-              rescue InitializationLeaseLost
-                break
-              rescue *CONNECTION_ERRORS => e
-                warn "Failed to renew queue initialization lease: #{e.message}"
-              end
-            end
-          end
-
-          executables = reorder_tests(tests, random: @ordering_random.dup)
-          chunks, individual_tests = executables.partition { |e| e.is_a?(CI::Queue::TestChunk) }
-          store_chunk_metadata(chunks) if chunks.any?
-          push(chunks.map(&:id) + individual_tests.map(&:id))
-        rescue InitializationLeaseLost
-          # Another process took over. Join its queue instead of publishing ours.
-          @master = false
-        ensure
-          if renewal
-            stop_renewal.set
-            renewal.join
-            release_initialization_lease
-          end
-        end
-
-        def master_lease_milliseconds
-          # Leave time for a surviving worker to take over within its setup wait.
-          [(config.queue_init_timeout.to_f * 500).ceil, 1].max
-        end
-
-        def with_master_lease
-          redis.watch(key('master-lease'), key('master-status')) do
-            unless redis.get(key('master-lease')) == @master_token && master_status == 'setup'
-              raise InitializationLeaseLost
-            end
-
-            result = redis.multi { |transaction| yield transaction }
-            raise InitializationLeaseLost unless result
-
-            result
-          end
-        end
-
-        def release_initialization_lease
-          with_master_lease do |transaction|
-            transaction.del(key('master-lease'), key('master-status'), key('master-worker-id'))
-          end
-          @master = false
-        rescue InitializationLeaseLost, *CONNECTION_ERRORS
-          # Published queues and replacement owners must remain untouched.
-        end
-
         # Runs a block while sending periodic heartbeats in a background thread.
         # This prevents other workers from stealing the test while it's being executed.
         def with_heartbeat(test_id)
@@ -503,21 +439,40 @@ def try_to_reserve_lost_test
           lost_test
         end
 
-        def push(tests)
+        def initialize_queue
+          return unless acquire_master_role?
+
+          executables = reorder_tests(index.values, random: @random)
+          chunks, individual_tests = executables.partition { |e| e.is_a?(CI::Queue::TestChunk) }
+          push(chunks.map(&:id) + individual_tests.map(&:id), chunks: chunks)
+        end
+
+        # Publishes the queue, its chunk metadata, and the ready status in one
+        # transaction. The setup marker can expire while a slow master is still
+        # ordering tests, so WATCH ensures only the first master to finish
+        # publishes; any other master joins that queue instead.
+        def push(tests, chunks: [])
           @total = tests.size
+          return unless @master
 
-          if @master
-            with_master_lease 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)
-              transaction.del(key('master-lease'))
+          published = redis.watch(key('master-status')) do
+            if queue_initialized?
+              redis.unwatch
+              nil
+            else
+              redis.multi do |transaction|
+                store_chunk_metadata(transaction, chunks)
+                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
           end
+          @master = false unless published
         rescue *CONNECTION_ERRORS
           raise if @master
         end
@@ -527,66 +482,65 @@ def register
         end
 
         def acquire_master_role?
-          @master_token = SecureRandom.hex(16)
-          @master = !!redis.watch(key('master-status'), key('master-lease')) do
-            if %w[ready finished].include?(master_status) || redis.get(key('master-lease'))
-              redis.unwatch
-              false
-            else
-              redis.multi do |transaction|
-                transaction.set(key('master-lease'), @master_token, px: master_lease_milliseconds)
-                transaction.set(key('master-status'), 'setup')
-                transaction.set(key('master-worker-id'), worker_id)
-                transaction.expire(key('master-worker-id'), config.redis_ttl)
-              end
+          return true if @master
+
+          @master = redis.set(key('master-status'), 'setup', nx: true, px: setup_timeout_milliseconds)
+          if @master
+            begin
+              redis.set(key('master-worker-id'), worker_id)
+              redis.expire(key('master-worker-id'), config.redis_ttl)
+              warn "Worker #{worker_id} elected as master"
+            rescue *CONNECTION_ERRORS
+              # If setting master-worker-id fails, we still have master status
+              # Log but don't lose master role
+              warn("Failed to set master-worker-id: #{$!.message}")
             end
           end
-          warn "Worker #{worker_id} elected as master" if @master
           @master
         rescue *CONNECTION_ERRORS
           @master = nil
           false
         end
 
+        def setup_timeout_milliseconds
+          # Expire early enough for a waiting worker to take over and still
+          # publish within its own queue_init_timeout.
+          [(config.queue_init_timeout.to_f * 500).ceil, 1].max
+        end
+
         def register_worker_presence
           register
           redis.expire(key('workers'), config.redis_ttl)
         rescue *CONNECTION_ERRORS
           raise if master?
         end
 
-        def store_chunk_metadata(chunks)
-          # Batch operations to avoid exceeding Redis multi operation limits
-          # Each chunk requires 4 commands (set, expire, sadd, hset), so batch conservatively
-          batch_size = 5 # 5 chunks = 20 commands + 2 expires = 22 commands per batch
-
-          chunks.each_slice(batch_size) do |chunk_batch|
-            with_master_lease do |transaction|
-              chunk_batch.each do |chunk|
-                # Store chunk metadata with TTL
-                transaction.set(
-                  key('chunk', chunk.id),
-                  chunk.to_json
-                )
-                transaction.expire(key('chunk', chunk.id), config.redis_ttl)
-
-                # Track all chunks for cleanup
-                transaction.sadd(key('chunks'), chunk.id)
-
-                # Store dynamic timeout for this chunk
-                # Timeout = estimated_duration (in ms) converted to seconds + buffer
-                # estimated_duration is in milliseconds, convert to seconds and add 10% buffer
-                buffer_percent = 10
-                estimated_duration_seconds = chunk.estimated_duration / 1000.0
-                chunk_timeout = (estimated_duration_seconds * (1 + buffer_percent / 100.0)).round(2)
-                # Format to string to avoid floating point precision issues in Redis
-                # Use %g to remove trailing zeros
-                transaction.hset(key('test-group-timeout'), chunk.id, format('%g', chunk_timeout))
-              end
-              transaction.expire(key('chunks'), config.redis_ttl)
-              transaction.expire(key('test-group-timeout'), config.redis_ttl)
-            end
+        def store_chunk_metadata(transaction, chunks)
+          return if chunks.empty?
+
+          chunks.each do |chunk|
+            # Store chunk metadata with TTL
+            transaction.set(
+              key('chunk', chunk.id),
+              chunk.to_json
+            )
+            transaction.expire(key('chunk', chunk.id), config.redis_ttl)
+
+            # Track all chunks for cleanup
+            transaction.sadd(key('chunks'), chunk.id)
+
+            # Store dynamic timeout for this chunk
+            # Timeout = estimated_duration (in ms) converted to seconds + buffer
+            # estimated_duration is in milliseconds, convert to seconds and add 10% buffer
+            buffer_percent = 10
+            estimated_duration_seconds = chunk.estimated_duration / 1000.0
+            chunk_timeout = (estimated_duration_seconds * (1 + buffer_percent / 100.0)).round(2)
+            # Format to string to avoid floating point precision issues in Redis
+            # Use %g to remove trailing zeros
+            transaction.hset(key('test-group-timeout'), chunk.id, format('%g', chunk_timeout))
           end
+          transaction.expire(key('chunks'), config.redis_ttl)
+          transaction.expire(key('test-group-timeout'), config.redis_ttl)
         end
 
         def chunk_id?(id)