[Shopify Sync] Use Concurrent::Set for thread-safety

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

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 require 'ci/queue/static'
-require 'set'
+require 'concurrent/set'
 
 module CI
   module Queue
@@ -21,6 +21,7 @@ class Worker < Base
 
         def initialize(redis, config)
           @reserved_test = nil
+          @reserved_tests = Concurrent::Set.new
           @shutdown_required = false
           @idle_since = nil
           super(redis, config)
@@ -301,6 +302,10 @@ def ensure_connection_and_script(script)
           redis.ping
         end
 
+        def reserved_tests
+          @reserved_tests ||= Concurrent::Set.new
+        end
+
         def worker_id
           config.worker_id
         end

ruby/lib/ci/queue/static.rb

@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require 'concurrent/set'
+
 module CI
   module Queue
     class Static
@@ -112,6 +114,10 @@ def should_requeue?(key)
       def requeues
         @requeues ||= Hash.new(0)
       end
+
+      def reserved_tests
+        @reserved_tests ||= Concurrent::Set.new
+      end
     end
   end
 end

ruby/test/integration/minitest_redis_test.rb

@@ -2,6 +2,7 @@
 require 'test_helper'
 require 'tmpdir'
 require 'active_support/testing/time_helpers'
+require 'concurrent/set'
 
 module Integration
   class MinitestRedisTest < Minitest::Test
@@ -308,6 +309,7 @@ def test_retry_report
       assert_equal 100, error_reports.size
 
       error_reports.keys.each_with_index do |test_id, index|
+        queue.instance_variable_set(:@reserved_tests, Concurrent::Set.new([test_id]))
         queue.build.record_success(test_id.dup, stats: {
           'assertions' => index + 1,
           'errors' => 0,

ruby/test/minitest/queue/build_status_recorder_test.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 require 'test_helper'
+require 'concurrent/set'
 
 module Minitest::Queue
   class BuildStatusRecorderTest < Minitest::Test
@@ -50,6 +51,10 @@ def test_retrying_test
 
     private
 
+    def reserve(queue, method_name)
+      queue.instance_variable_set(:@reserved_tests, Concurrent::Set.new([Minitest::Queue::SingleExample.new("Minitest::Test", method_name).id]))
+    end
+
     def worker(id)
       CI::Queue::Redis.new(
         @redis_url,