From 8b3da9dfc06a77ebe18a5d0558d3700de204d3b9 Mon Sep 17 00:00:00 2001
From: Clinton Kwarteng <ckwarteng@figma.com>
Date: Fri, 9 Oct 2026 23:14:54 +0000
Subject: [PATCH 1/3] ruby: Recover queue initialization when the master exits

---
 ruby/lib/ci/queue/redis/base.rb               |   3 +
 ruby/lib/ci/queue/redis/worker.rb             | 112 +++++++++----
 .../ci/queue/redis_initialization_test.rb     | 152 ++++++++++++++++++
 3 files changed, 240 insertions(+), 27 deletions(-)
 create mode 100644 ruby/test/ci/queue/redis_initialization_test.rb

diff --git a/ruby/lib/ci/queue/redis/base.rb b/ruby/lib/ci/queue/redis/base.rb
index 9ebdd58..335b8f7 100644
--- a/ruby/lib/ci/queue/redis/base.rb
+++ b/ruby/lib/ci/queue/redis/base.rb
@@ -59,6 +59,9 @@ def wait_for_master(timeout: 120)
           (timeout * 10 + 1).to_i.times do
             return true if queue_initialized?
 
+            yield if block_given?
+            return true if queue_initialized?
+
             sleep 0.1
           end
           raise LostMaster, "The master worker (worker #{master_worker_id}) is still `#{master_status}` after #{timeout} seconds waiting."
diff --git a/ruby/lib/ci/queue/redis/worker.rb b/ruby/lib/ci/queue/redis/worker.rb
index 7338a45..9d3d9f8 100644
--- a/ruby/lib/ci/queue/redis/worker.rb
+++ b/ruby/lib/ci/queue/redis/worker.rb
@@ -1,7 +1,9 @@
 # frozen_string_literal: true
 
 require 'ci/queue/static'
+require 'concurrent/atomic/event'
 require 'concurrent/set'
+require 'securerandom'
 
 module CI
   module Queue
@@ -17,6 +19,8 @@ class << self
 
       class Worker < Base
         DEFAULT_SLEEP_SECONDS = 0.5
+        InitializationLeaseLost = Class.new(StandardError)
+        private_constant :InitializationLeaseLost
         attr_reader :total
 
         def initialize(redis, config)
@@ -37,18 +41,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
-
-          if acquire_master_role?
-            executables = reorder_tests(tests, random: random)
-
-            chunks = executables.select { |e| e.is_a?(CI::Queue::TestChunk) }
-            individual_tests = executables.reject { |e| e.is_a?(CI::Queue::TestChunk) }
-
-            store_chunk_metadata(chunks) if chunks.any?
-
-            all_ids = chunks.map(&:id) + individual_tests.map(&:id)
-            push(all_ids)
-          end
+          @ordering_random = random.dup
+          initialize_queue(tests)
 
           register_worker_presence
 
@@ -76,7 +70,9 @@ def idle?
         end
 
         def poll
-          wait_for_master
+          wait_for_master(timeout: config.queue_init_timeout) do
+            initialize_queue(index.values)
+          end
           if master?
             warn "Worker #{worker_id} is the master"
           else
@@ -274,6 +270,66 @@ 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)
@@ -451,7 +507,7 @@ def push(tests)
           @total = tests.size
 
           if @master
-            redis.multi do |transaction|
+            with_master_lease do |transaction|
               transaction.lpush(key('queue'), tests) unless tests.empty?
               transaction.set(key('total'), @total)
               transaction.set(key('master-status'), 'ready')
@@ -459,6 +515,7 @@ def push(tests)
               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'))
             end
           end
         rescue *CONNECTION_ERRORS
@@ -470,20 +527,21 @@ def register
         end
 
         def acquire_master_role?
-          return true if @master
-
-          @master = redis.setnx(key('master-status'), 'setup')
-          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}")
+          @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
             end
           end
+          warn "Worker #{worker_id} elected as master" if @master
           @master
         rescue *CONNECTION_ERRORS
           @master = nil
@@ -503,7 +561,7 @@ def store_chunk_metadata(chunks)
           batch_size = 5 # 5 chunks = 20 commands + 2 expires = 22 commands per batch
 
           chunks.each_slice(batch_size) do |chunk_batch|
-            redis.multi do |transaction|
+            with_master_lease do |transaction|
               chunk_batch.each do |chunk|
                 # Store chunk metadata with TTL
                 transaction.set(
diff --git a/ruby/test/ci/queue/redis_initialization_test.rb b/ruby/test/ci/queue/redis_initialization_test.rb
new file mode 100644
index 0000000..fe876cd
--- /dev/null
+++ b/ruby/test/ci/queue/redis_initialization_test.rb
@@ -0,0 +1,152 @@
+# frozen_string_literal: true
+
+require 'test_helper'
+
+class CI::Queue::Redis::InitializationTest < Minitest::Test
+  include QueueHelper
+
+  def setup
+    @redis_url = ENV.fetch('REDIS_URL', 'redis://localhost:6379/0')
+    @redis = ::Redis.new(url: @redis_url)
+    @redis.flushdb
+  end
+
+  def teardown
+    @redis.flushdb
+  end
+
+  def test_surviving_worker_recovers_when_initializer_exits
+    child = fork do
+      initializer = worker(0)
+      initializer.define_singleton_method(:reorder_tests) { |*| exit! 143 }
+      initializer.populate(tests)
+    end
+    _, status = Process.wait2(child)
+
+    assert_equal 143, status.exitstatus
+    assert_equal 'setup', @redis.get(key('master-status'))
+    assert_operator @redis.pttl(key('master-lease')), :>, 0
+
+    survivor = worker(1).populate(tests)
+    refute_predicate survivor, :master?
+    assert_equal tests.sort, poll(survivor).sort
+    assert_predicate survivor, :master?
+    assert_equal '1', @redis.get(key('master-worker-id'))
+    assert_equal 'ready', @redis.get(key('master-status'))
+    assert_nil @redis.get(key('master-lease'))
+  end
+
+  def test_worker_recovers_a_setup_marker_without_a_lease
+    @redis.set(key('master-status'), 'setup')
+    @redis.set(key('master-worker-id'), '0')
+
+    replacement = worker(1).populate(tests)
+
+    assert_predicate replacement, :master?
+    assert_equal tests.sort, poll(replacement).sort
+    assert_equal '1', @redis.get(key('master-worker-id'))
+  end
+
+  def test_setup_exception_releases_election
+    initializer = worker(0)
+    initializer.define_singleton_method(:reorder_tests) { |*| raise 'setup failed' }
+
+    error = assert_raises(RuntimeError) { initializer.populate(tests) }
+    assert_equal 'setup failed', error.message
+
+    replacement = worker(1).populate(tests)
+    assert_predicate replacement, :master?
+    assert_equal tests.sort, poll(replacement).sort
+  end
+
+  def test_late_initializer_cannot_publish_over_replacement
+    initializer = worker(0)
+    replacement = worker(1)
+    redis = @redis
+    lease_key = key('master-lease')
+    initializer.define_singleton_method(:reorder_tests) do |input, **|
+      redis.del(lease_key)
+      replacement.populate(input)
+      input.reverse
+    end
+    initializer.populate(tests)
+
+    assert_predicate replacement, :master?
+    refute_predicate initializer, :master?
+    assert_equal tests.size, @redis.llen(key('queue'))
+    assert_equal tests.sort, poll(replacement).sort
+    assert_equal '1', @redis.get(key('master-worker-id'))
+  end
+
+  def test_late_initializer_cannot_overwrite_chunk_metadata
+    initializer = worker(0)
+    replacement = worker(1, strategy: :suite_bin_packing)
+    redis = @redis
+    lease_key = key('master-lease')
+    chunk_key = key('chunk:ATest:chunk_0')
+    published_metadata = nil
+    initializer.define_singleton_method(:reorder_tests) do |input, **|
+      redis.del(lease_key)
+      replacement.populate(input)
+      published_metadata = redis.get(chunk_key)
+      [CI::Queue::TestChunk.new('ATest:chunk_0', 'ATest', [input.first.id], 9999)]
+    end
+    initializer.populate(tests)
+
+    refute_nil published_metadata
+    assert_equal published_metadata, @redis.get(chunk_key)
+    refute_predicate initializer, :master?
+    assert_equal tests.sort, poll(replacement).flat_map(&:tests).sort
+  end
+
+  def test_live_initializer_renews_lease_during_slow_setup
+    initializer = worker(0)
+    started = Queue.new
+    proceed = Queue.new
+    initializer.define_singleton_method(:reorder_tests) do |input, **|
+      started << true
+      proceed.pop
+      input
+    end
+    thread = Thread.new { initializer.populate(tests) }
+    started.pop
+    sleep 0.4 # Longer than the initial 0.25-second lease.
+
+    contender = worker(1).populate(tests)
+    refute_predicate contender, :master?
+    assert_equal '0', @redis.get(key('master-worker-id'))
+
+    proceed << true
+    assert thread.join(2), 'initializer did not complete'
+    assert_predicate initializer, :master?
+    assert_equal tests.sort, poll(contender).sort
+  ensure
+    proceed << true if proceed
+    thread&.kill
+    thread&.join
+  end
+
+  private
+
+  def tests
+    SharedQueueAssertions::TEST_LIST.dup
+  end
+
+  def key(name)
+    "build:initialization:#{name}"
+  end
+
+  def worker(id, **options)
+    CI::Queue::Redis.new(
+      @redis_url,
+      CI::Queue::Configuration.new(
+        build_id: 'initialization',
+        worker_id: id.to_s,
+        timeout: 0.5,
+        queue_init_timeout: 0.5,
+        heartbeat_interval: 0,
+        **options,
+      ),
+    )
+  end
+end

From 9d7fb11f098999b08ab7f938d17a05e9960cbb8a Mon Sep 17 00:00:00 2001
From: Clinton Kwarteng <ckwarteng@figma.com>
Date: Sat, 10 Oct 2026 00:19:46 +0000
Subject: [PATCH 2/3] ruby: Simplify LostMaster recovery to an expiring setup
 marker

Replace the renewable initialization lease with two smaller pieces:

- The setup marker expires after half of queue_init_timeout, so a waiting
  worker can take over when the elected master dies before publishing.
- Publication (queue, chunk metadata, total, ready status) is one
  WATCH-guarded transaction, so if a slow master and its replacement both
  finish, only the first publishes and the other joins that queue.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 ruby/lib/ci/queue/redis/worker.rb             | 197 +++++++-----------
 .../ci/queue/redis_initialization_test.rb     |  57 ++---
 2 files changed, 102 insertions(+), 152 deletions(-)

diff --git a/ruby/lib/ci/queue/redis/worker.rb b/ruby/lib/ci/queue/redis/worker.rb
index 9d3d9f8..dc41b98 100644
--- a/ruby/lib/ci/queue/redis/worker.rb
+++ b/ruby/lib/ci/queue/redis/worker.rb
@@ -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,7 @@ def idle?
         end
 
         def poll
-          wait_for_master(timeout: config.queue_init_timeout) do
-            initialize_queue(index.values)
-          end
+          wait_for_master(timeout: config.queue_init_timeout) { initialize_queue }
           if master?
             warn "Worker #{worker_id} is the master"
           else
@@ -270,66 +264,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 +437,38 @@ 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
+
+        # The setup marker can expire under a slow master, so only the first
+        # master to finish publishes.
+        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,27 +478,31 @@ 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
+          # Leave a waiting worker time to take over within 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)
@@ -555,38 +510,32 @@ def register_worker_presence
           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)
diff --git a/ruby/test/ci/queue/redis_initialization_test.rb b/ruby/test/ci/queue/redis_initialization_test.rb
index fe876cd..1a977c3 100644
--- a/ruby/test/ci/queue/redis_initialization_test.rb
+++ b/ruby/test/ci/queue/redis_initialization_test.rb
@@ -16,16 +16,7 @@ def teardown
   end
 
   def test_surviving_worker_recovers_when_initializer_exits
-    child = fork do
-      initializer = worker(0)
-      initializer.define_singleton_method(:reorder_tests) { |*| exit! 143 }
-      initializer.populate(tests)
-    end
-    _, status = Process.wait2(child)
-
-    assert_equal 143, status.exitstatus
-    assert_equal 'setup', @redis.get(key('master-status'))
-    assert_operator @redis.pttl(key('master-lease')), :>, 0
+    exit_during_setup(worker_id: 0)
 
     survivor = worker(1).populate(tests)
     refute_predicate survivor, :master?
@@ -33,21 +24,18 @@ def test_surviving_worker_recovers_when_initializer_exits
     assert_predicate survivor, :master?
     assert_equal '1', @redis.get(key('master-worker-id'))
     assert_equal 'ready', @redis.get(key('master-status'))
-    assert_nil @redis.get(key('master-lease'))
   end
 
-  def test_worker_recovers_a_setup_marker_without_a_lease
-    @redis.set(key('master-status'), 'setup')
-    @redis.set(key('master-worker-id'), '0')
-
-    replacement = worker(1).populate(tests)
+  def test_retried_initializer_recovers_its_own_setup_marker
+    exit_during_setup(worker_id: 0)
 
-    assert_predicate replacement, :master?
-    assert_equal tests.sort, poll(replacement).sort
-    assert_equal '1', @redis.get(key('master-worker-id'))
+    retried = worker(0).populate(tests)
+    assert_equal tests.sort, poll(retried).sort
+    assert_predicate retried, :master?
+    assert_equal 'ready', @redis.get(key('master-status'))
   end
 
-  def test_setup_exception_releases_election
+  def test_worker_takes_over_after_setup_exception
     initializer = worker(0)
     initializer.define_singleton_method(:reorder_tests) { |*| raise 'setup failed' }
 
@@ -55,17 +43,17 @@ def test_setup_exception_releases_election
     assert_equal 'setup failed', error.message
 
     replacement = worker(1).populate(tests)
-    assert_predicate replacement, :master?
     assert_equal tests.sort, poll(replacement).sort
+    assert_predicate replacement, :master?
   end
 
   def test_late_initializer_cannot_publish_over_replacement
     initializer = worker(0)
     replacement = worker(1)
     redis = @redis
-    lease_key = key('master-lease')
+    status_key = key('master-status')
     initializer.define_singleton_method(:reorder_tests) do |input, **|
-      redis.del(lease_key)
+      redis.del(status_key)
       replacement.populate(input)
       input.reverse
     end
@@ -82,11 +70,11 @@ def test_late_initializer_cannot_overwrite_chunk_metadata
     initializer = worker(0)
     replacement = worker(1, strategy: :suite_bin_packing)
     redis = @redis
-    lease_key = key('master-lease')
+    status_key = key('master-status')
     chunk_key = key('chunk:ATest:chunk_0')
     published_metadata = nil
     initializer.define_singleton_method(:reorder_tests) do |input, **|
-      redis.del(lease_key)
+      redis.del(status_key)
       replacement.populate(input)
       published_metadata = redis.get(chunk_key)
       [CI::Queue::TestChunk.new('ATest:chunk_0', 'ATest', [input.first.id], 9999)]
@@ -99,7 +87,7 @@ def test_late_initializer_cannot_overwrite_chunk_metadata
     assert_equal tests.sort, poll(replacement).flat_map(&:tests).sort
   end
 
-  def test_live_initializer_renews_lease_during_slow_setup
+  def test_waiting_worker_does_not_take_over_a_live_setup
     initializer = worker(0)
     started = Queue.new
     proceed = Queue.new
@@ -110,16 +98,16 @@ def test_live_initializer_renews_lease_during_slow_setup
     end
     thread = Thread.new { initializer.populate(tests) }
     started.pop
-    sleep 0.4 # Longer than the initial 0.25-second lease.
 
     contender = worker(1).populate(tests)
     refute_predicate contender, :master?
-    assert_equal '0', @redis.get(key('master-worker-id'))
+    assert_equal 'setup', @redis.get(key('master-status'))
 
     proceed << true
     assert thread.join(2), 'initializer did not complete'
     assert_predicate initializer, :master?
     assert_equal tests.sort, poll(contender).sort
+    refute_predicate contender, :master?
   ensure
     proceed << true if proceed
     thread&.kill
@@ -128,6 +116,19 @@ def test_live_initializer_renews_lease_during_slow_setup
 
   private
 
+  def exit_during_setup(worker_id:)
+    child = fork do
+      initializer = worker(worker_id)
+      initializer.define_singleton_method(:reorder_tests) { |*| exit! 143 }
+      initializer.populate(tests)
+    end
+    _, status = Process.wait2(child)
+
+    assert_equal 143, status.exitstatus
+    assert_equal 'setup', @redis.get(key('master-status'))
+    assert_operator @redis.pttl(key('master-status')), :>, 0
+  end
+
   def tests
     SharedQueueAssertions::TEST_LIST.dup
   end

From 263461ca8cf223059a1beaafaecaecc9834b3b55 Mon Sep 17 00:00:00 2001
From: Clinton Kwarteng <ckwarteng@figma.com>
Date: Sun, 11 Oct 2026 23:31:48 +0000
Subject: [PATCH 3/3] ruby: Stage initializer chunks in bounded transactions

---
 ruby/lib/ci/queue/redis/worker.rb             | 71 +++++++++++--------
 .../ci/queue/redis/dynamic_timeout_test.rb    | 24 ++++---
 ruby/test/ci/queue/redis/worker_chunk_test.rb | 28 ++++----
 .../ci/queue/redis_initialization_test.rb     | 46 +++++++++++-
 ruby/test/ci/queue/redis_test.rb              |  8 +--
 5 files changed, 122 insertions(+), 55 deletions(-)

diff --git a/ruby/lib/ci/queue/redis/worker.rb b/ruby/lib/ci/queue/redis/worker.rb
index dc41b98..f460f5c 100644
--- a/ruby/lib/ci/queue/redis/worker.rb
+++ b/ruby/lib/ci/queue/redis/worker.rb
@@ -2,6 +2,7 @@
 
 require 'ci/queue/static'
 require 'concurrent/set'
+require 'securerandom'
 
 module CI
   module Queue
@@ -442,12 +443,22 @@ def initialize_queue
 
           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)
+          # An initializer may outlive its setup marker. Give its chunks unique
+          # IDs so staging metadata cannot overwrite another initializer's data.
+          initialization_id = SecureRandom.hex(8)
+          chunks = chunks.map do |chunk|
+            CI::Queue::TestChunk.new(
+              "#{chunk.id}:#{initialization_id}", chunk.suite_name,
+              chunk.test_ids, chunk.estimated_duration, test_count: chunk.test_count
+            )
+          end
+          store_chunk_metadata(chunks)
+          push(chunks.map(&:id) + individual_tests.map(&:id))
         end
 
         # The setup marker can expire under a slow master, so only the first
         # master to finish publishes.
-        def push(tests, chunks: [])
+        def push(tests)
           @total = tests.size
           return unless @master
 
@@ -457,7 +468,6 @@ def push(tests, chunks: [])
               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')
@@ -510,32 +520,37 @@ def register_worker_presence
           raise if master?
         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))
+        def store_chunk_metadata(chunks)
+          # Four commands per chunk, plus two expiry commands: at most 22.
+          batch_size = 5
+
+          chunks.each_slice(batch_size) do |chunk_batch|
+            redis.multi 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
           end
-          transaction.expire(key('chunks'), config.redis_ttl)
-          transaction.expire(key('test-group-timeout'), config.redis_ttl)
         end
 
         def chunk_id?(id)
diff --git a/ruby/test/ci/queue/redis/dynamic_timeout_test.rb b/ruby/test/ci/queue/redis/dynamic_timeout_test.rb
index 4fd3816..985287f 100644
--- a/ruby/test/ci/queue/redis/dynamic_timeout_test.rb
+++ b/ruby/test/ci/queue/redis/dynamic_timeout_test.rb
@@ -38,7 +38,7 @@ def test_chunk_timeout_stored_in_redis_hash
 
     # Verify timeout was stored in test-group-timeout hash
     # Timeout should be: estimated_duration (5000ms = 5s) * 1.1 buffer = 5.5s
-    chunk_timeout = @redis.hget('build:42:test-group-timeout', 'TestA:chunk_0')
+    chunk_timeout = @redis.hget('build:42:test-group-timeout', stored_chunk_id('TestA:chunk_0'))
     refute_nil chunk_timeout
     assert_equal '5.5', chunk_timeout
   end
@@ -54,7 +54,7 @@ def test_chunk_timeout_scales_with_test_count
       worker.populate(small_tests)
     end
 
-    small_timeout = @redis.hget('build:42:test-group-timeout', 'SmallSuite:chunk_0')
+    small_timeout = @redis.hget('build:42:test-group-timeout', stored_chunk_id('SmallSuite:chunk_0'))
     assert_equal '1.1', small_timeout # 1000ms = 1s * 1.1 buffer = 1.1s
 
     @redis.flushdb
@@ -69,7 +69,7 @@ def test_chunk_timeout_scales_with_test_count
       worker.populate(large_tests)
     end
 
-    large_timeout = @redis.hget('build:42:test-group-timeout', 'LargeSuite:chunk_0')
+    large_timeout = @redis.hget('build:42:test-group-timeout', stored_chunk_id('LargeSuite:chunk_0'))
     assert_equal '5.5', large_timeout # 5000ms = 5s * 1.1 buffer = 5.5s
   end
 
@@ -92,9 +92,9 @@ def test_multiple_chunks_stored_with_different_timeouts
     end
 
     # Verify each chunk has correct timeout
-    assert_equal '2.2', @redis.hget('build:42:test-group-timeout', 'TestA:chunk_0') # 2000ms = 2s * 1.1 buffer = 2.2s
-    assert_equal '3.3', @redis.hget('build:42:test-group-timeout', 'TestB:chunk_0') # 3000ms = 3s * 1.1 buffer = 3.3s
-    assert_equal '1.1', @redis.hget('build:42:test-group-timeout', 'TestC:chunk_0') # 1000ms = 1s * 1.1 buffer = 1.1s
+    assert_equal '2.2', @redis.hget('build:42:test-group-timeout', stored_chunk_id('TestA:chunk_0')) # 2000ms = 2s * 1.1 buffer = 2.2s
+    assert_equal '3.3', @redis.hget('build:42:test-group-timeout', stored_chunk_id('TestB:chunk_0')) # 3000ms = 3s * 1.1 buffer = 3.3s
+    assert_equal '1.1', @redis.hget('build:42:test-group-timeout', stored_chunk_id('TestC:chunk_0')) # 1000ms = 1s * 1.1 buffer = 1.1s
   end
 
   def test_timeout_hash_has_ttl
@@ -145,8 +145,8 @@ def test_mixed_chunks_and_tests_only_chunks_have_timeouts
     end
 
     # Chunks should have timeouts
-    assert_equal '2.2', @redis.hget('build:42:test-group-timeout', 'TestA:chunk_0') # 2000ms = 2s * 1.1 buffer = 2.2s
-    assert_equal '3.3', @redis.hget('build:42:test-group-timeout', 'TestC:chunk_0') # 3000ms = 3s * 1.1 buffer = 3.3s
+    assert_equal '2.2', @redis.hget('build:42:test-group-timeout', stored_chunk_id('TestA:chunk_0')) # 2000ms = 2s * 1.1 buffer = 2.2s
+    assert_equal '3.3', @redis.hget('build:42:test-group-timeout', stored_chunk_id('TestC:chunk_0')) # 3000ms = 3s * 1.1 buffer = 3.3s
 
     # Individual test should not
     assert_nil @redis.hget('build:42:test-group-timeout', 'TestB#test_1')
@@ -235,7 +235,7 @@ def test_chunk_not_marked_lost_before_dynamic_timeout
 
     # Reserve the chunk with worker1
     reserved_id = worker1.send(:try_to_reserve_test)
-    assert_equal 'TestSuite:chunk_0', reserved_id
+    assert_match /\ATestSuite:chunk_0:[0-9a-f]{16}\z/, reserved_id
 
     # Wait 1 second (less than 2.5s timeout)
     sleep 1
@@ -296,7 +296,7 @@ def test_batching_with_many_chunks
 
     # Verify all chunks have timeouts stored despite batching
     chunks.each do |chunk|
-      timeout = @redis.hget('build:42:test-group-timeout', chunk.id)
+      timeout = @redis.hget('build:42:test-group-timeout', stored_chunk_id(chunk.id))
       refute_nil timeout, "Chunk #{chunk.id} should have timeout stored"
       assert_equal '1.1', timeout # 1000ms = 1s * 1.1 buffer = 1.1s
     end
@@ -304,6 +304,10 @@ def test_batching_with_many_chunks
 
   private
 
+  def stored_chunk_id(original_id)
+    @redis.smembers('build:42:chunks').find { |id| id.start_with?("#{original_id}:") }
+  end
+
   def create_mock_tests(test_ids)
     test_ids.map { |id| MockTest.new(id) }
   end
diff --git a/ruby/test/ci/queue/redis/worker_chunk_test.rb b/ruby/test/ci/queue/redis/worker_chunk_test.rb
index 95b8a10..7aba38d 100644
--- a/ruby/test/ci/queue/redis/worker_chunk_test.rb
+++ b/ruby/test/ci/queue/redis/worker_chunk_test.rb
@@ -36,7 +36,7 @@ def test_populate_stores_chunk_metadata_in_redis
     end
 
     # Verify chunk metadata was stored
-    chunk_data = @redis.get('build:42:chunk:TestA:chunk_0')
+    chunk_data = @redis.get("build:42:chunk:#{stored_chunk_id('TestA:chunk_0')}")
     refute_nil chunk_data
 
     parsed = JSON.parse(chunk_data)
@@ -100,10 +100,10 @@ def test_resolve_chunk
     end
 
     # Manually store and resolve chunk
-    resolved = @worker.send(:resolve_chunk, 'TestA:chunk_0')
+    resolved = @worker.send(:resolve_chunk, stored_chunk_id('TestA:chunk_0'))
 
     assert_instance_of CI::Queue::Redis::Worker::ResolvedChunk, resolved
-    assert_equal 'TestA:chunk_0', resolved.id
+    assert_match /\ATestA:chunk_0(?::[0-9a-f]{16})?\z/, resolved.id
     assert_equal 'TestA', resolved.suite_name
     assert_equal 3, resolved.tests.size
     assert_equal test_ids, resolved.tests.map(&:id)
@@ -120,10 +120,10 @@ def test_resolve_chunk_with_subset_of_tests
       @worker.populate(tests)
     end
 
-    resolved = @worker.send(:resolve_chunk, 'TestA:chunk_0')
+    resolved = @worker.send(:resolve_chunk, stored_chunk_id('TestA:chunk_0'))
 
     assert_instance_of CI::Queue::Redis::Worker::ResolvedChunk, resolved
-    assert_equal 'TestA:chunk_0', resolved.id
+    assert_match /\ATestA:chunk_0(?::[0-9a-f]{16})?\z/, resolved.id
     assert_equal 2, resolved.tests.size
     assert_equal ['TestA#test_1', 'TestA#test_3'], resolved.tests.map(&:id)
   end
@@ -140,7 +140,7 @@ def test_resolved_chunk_interface
 
     resolved = CI::Queue::Redis::Worker::ResolvedChunk.new(chunk, tests)
 
-    assert_equal 'TestA:chunk_0', resolved.id
+    assert_match /\ATestA:chunk_0(?::[0-9a-f]{16})?\z/, resolved.id
     assert_equal 'TestA', resolved.suite_name
     assert_equal 2, resolved.size
     assert resolved.chunk?
@@ -195,7 +195,7 @@ def test_populate_with_mixed_chunks_and_tests
 
     # Check that both chunk and individual test IDs are in queue
     queue_items = @redis.lrange('build:42:queue', 0, -1)
-    assert_includes queue_items, 'TestA:chunk_0'
+    assert_includes queue_items, stored_chunk_id('TestA:chunk_0')
     assert_includes queue_items, 'TestB#test_1'
   end
 
@@ -210,12 +210,12 @@ def test_chunk_metadata_has_ttl
       @worker.populate(tests)
     end
 
-    ttl = @redis.ttl('build:42:chunk:TestA:chunk_0')
+    ttl = @redis.ttl("build:42:chunk:#{stored_chunk_id('TestA:chunk_0')}")
     assert ttl > 0, 'Chunk metadata should have TTL set'
   end
 
   def test_populate_with_many_chunks_uses_batching
-    # Create 20 test chunks to verify batching works (batch size is 7)
+    # Create 20 test chunks to verify batching works (batch size is 5)
     tests = (1..20).map { |i| MockTest.new("TestSuite#{i}#test_1") }
     chunks = (1..20).map do |i|
       test_ids = ["TestSuite#{i}#test_1"]
@@ -228,7 +228,7 @@ def test_populate_with_many_chunks_uses_batching
 
     # Verify all chunks were stored despite batching
     chunks.each do |chunk|
-      chunk_data = @redis.get("build:42:chunk:#{chunk.id}")
+      chunk_data = @redis.get("build:42:chunk:#{stored_chunk_id(chunk.id)}")
       refute_nil chunk_data, "Chunk #{chunk.id} should be stored"
 
       parsed = JSON.parse(chunk_data)
@@ -239,7 +239,7 @@ def test_populate_with_many_chunks_uses_batching
     # Verify all chunk IDs are in the chunks set
     stored_chunks = @redis.smembers('build:42:chunks')
     chunks.each do |chunk|
-      assert_includes stored_chunks, chunk.id
+      assert_includes stored_chunks, stored_chunk_id(chunk.id)
     end
   end
 
@@ -257,7 +257,7 @@ def test_worker_history_retry_expands_chunks_and_deduplicates_tests
     end
 
     history_key = 'build:42:worker:1:queue'
-    [chunk.id, tests.last.id, chunk.id, tests[1].id].each do |id|
+    [stored_chunk_id(chunk.id), tests.last.id, stored_chunk_id(chunk.id), tests[1].id].each do |id|
       @redis.lpush(history_key, id)
     end
 
@@ -304,6 +304,10 @@ def test_worker_history_retry_rejects_invalid_chunk_metadata
 
   private
 
+  def stored_chunk_id(original_id)
+    @redis.smembers('build:42:chunks').find { |id| id.start_with?("#{original_id}:") }
+  end
+
   def create_mock_tests(test_ids)
     test_ids.map { |id| MockTest.new(id) }
   end
diff --git a/ruby/test/ci/queue/redis_initialization_test.rb b/ruby/test/ci/queue/redis_initialization_test.rb
index 1a977c3..8181456 100644
--- a/ruby/test/ci/queue/redis_initialization_test.rb
+++ b/ruby/test/ci/queue/redis_initialization_test.rb
@@ -71,11 +71,13 @@ def test_late_initializer_cannot_overwrite_chunk_metadata
     replacement = worker(1, strategy: :suite_bin_packing)
     redis = @redis
     status_key = key('master-status')
-    chunk_key = key('chunk:ATest:chunk_0')
+    chunk_key = nil
     published_metadata = nil
     initializer.define_singleton_method(:reorder_tests) do |input, **|
       redis.del(status_key)
       replacement.populate(input)
+      chunk_id = redis.lrange(key('queue'), 0, -1).find { |id| id.start_with?('ATest:chunk_0:') }
+      chunk_key = key("chunk:#{chunk_id}")
       published_metadata = redis.get(chunk_key)
       [CI::Queue::TestChunk.new('ATest:chunk_0', 'ATest', [input.first.id], 9999)]
     end
@@ -114,6 +116,48 @@ def test_waiting_worker_does_not_take_over_a_live_setup
     thread&.join
   end
 
+  def test_chunk_initialization_keeps_transactions_bounded
+    initializer = worker(0, strategy: :suite_bin_packing)
+    client = initializer.send(:redis)
+    original_multi = client.method(:multi)
+    transaction_sizes = []
+    client.define_singleton_method(:multi) do |*args, &block|
+      original_multi.call(*args) do |transaction|
+        count = 0
+        proxy = Object.new
+        %i[set expire sadd hset lpush llen zcard].each do |command|
+          proxy.define_singleton_method(command) do |*command_args|
+            count += 1
+            transaction.public_send(command, *command_args)
+          end
+        end
+        result = block.call(proxy)
+        raise ArgumentError, 'Maximum cardinality of 25 exceeded' if count > 25
+
+        transaction_sizes << count
+        result
+      end
+    end
+    input = 20.times.map { |i| SharedTestCases::TestCase.new("Suite#{i}#test") }
+
+    initializer.populate(input)
+    assert_equal input.sort, poll(initializer).flat_map(&:tests).sort
+    assert_equal [22, 22, 22, 22, 6], transaction_sizes.take(5)
+  end
+
+  def test_recovery_after_partial_chunk_staging
+    initializer = worker(0, strategy: :suite_bin_packing)
+    initializer.define_singleton_method(:store_chunk_metadata) do |chunks|
+      super(chunks.first(1))
+      raise 'staging failed'
+    end
+    assert_raises(RuntimeError) { initializer.populate(tests) }
+
+    replacement = worker(0, strategy: :suite_bin_packing).populate(tests)
+    assert_equal tests.sort, poll(replacement).flat_map(&:tests).sort
+    assert_predicate replacement, :master?
+  end
+
   private
 
   def exit_during_setup(worker_id:)
diff --git a/ruby/test/ci/queue/redis_test.rb b/ruby/test/ci/queue/redis_test.rb
index d50ec39..fd7b3ff 100644
--- a/ruby/test/ci/queue/redis_test.rb
+++ b/ruby/test/ci/queue/redis_test.rb
@@ -491,7 +491,7 @@ def test_suite_bin_packing_uses_moving_average_for_duration
     assert_equal 1, chunks.size
     chunk = chunks.first
     assert chunk.chunk?, 'Expected a chunk'
-    assert_equal 'TestSuite:chunk_0', chunk.id
+    assert_match /\ATestSuite:chunk_0:[0-9a-f]{16}\z/, chunk.id
     assert_equal 8000.0, chunk.estimated_duration
   end
 
@@ -635,11 +635,11 @@ def test_moving_average_ordering_by_duration
 
     # Should be ordered by duration descending: SlowTest, MediumTest, FastTest
     assert_equal 3, chunks.size
-    assert_equal 'SlowTest:chunk_0', chunks[0].id
+    assert_match /\ASlowTest:chunk_0:[0-9a-f]{16}\z/, chunks[0].id
     assert_equal 10_000.0, chunks[0].estimated_duration
-    assert_equal 'MediumTest:chunk_0', chunks[1].id
+    assert_match /\AMediumTest:chunk_0:[0-9a-f]{16}\z/, chunks[1].id
     assert_equal 5000.0, chunks[1].estimated_duration
-    assert_equal 'FastTest:chunk_0', chunks[2].id
+    assert_match /\AFastTest:chunk_0:[0-9a-f]{16}\z/, chunks[2].id
     assert_equal 1000.0, chunks[2].estimated_duration
   end