Fix race condition

There is a small race condition between acknowledge and recording the error
in which the reporter may see the queue empty and no error recorded
and wrongly thinks it's a success.

Author: ChrisBr

redis/acknowledge.lua

@@ -1,9 +1,18 @@
 local zset_key = KEYS[1]
 local processed_key = KEYS[2]
 local owners_key = KEYS[3]
+local error_reports_key = KEYS[4]
 
 local test = ARGV[1]
-
+local error = ARGV[2]
+local ttl = ARGV[3]
 redis.call('zrem', zset_key, test)
 redis.call('hdel', owners_key, test)  -- Doesn't matter if it was reclaimed by another workers
-return redis.call('sadd', processed_key, test)
+local acknowledged = redis.call('sadd', processed_key, test)
+
+if acknowledged and error ~= "" then
+  redis.call('hset', error_reports_key, test, error)
+  redis.call('expire', error_reports_key, ttl)
+end
+
+return acknowledged

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

@@ -181,8 +181,8 @@ def queue_initializing?
           master_status == 'setup'
         end
 
-        def increment_test_failed
-          redis.incr(key('test_failed_count'))
+        def increment_test_failed(pipeline: redis)
+          pipeline.incr(key('test_failed_count'))
         end
 
         def test_failed
@@ -226,7 +226,8 @@ def master_status
         end
 
         def eval_script(script, *args)
-          redis.evalsha(load_script(script), *args)
+          pipeline = args.delete(:pipeline) || redis
+          pipeline.evalsha(load_script(script), *args)
         end
 
         def load_script(script)

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

@@ -59,20 +59,10 @@ def record_warning(type, attributes)
         Test = Struct.new(:id) # Hack
 
         def record_error(id, payload, stats: nil)
-          # FIXME: the ack and hset should be atomic
-          # otherwise the reporter may see the empty queue before the error
-          # is appended and wrongly think it's a success.
-          if @queue.acknowledge(id)
-            redis.pipelined do |pipeline|
-              pipeline.hset(
-                key('error-reports'),
-                id,
-                payload,
-              )
-              pipeline.expire(key('error-reports'), config.redis_ttl)
-              record_stats(stats, pipeline: pipeline)
-              @queue.increment_test_failed
-            end
+          redis.pipelined do |pipeline|
+            @queue.acknowledge(id, error: payload, pipeline: pipeline)
+            record_stats(stats, pipeline: pipeline)
+            @queue.increment_test_failed(pipeline: pipeline)
           end
           nil
         end

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

@@ -97,12 +97,13 @@ def report_worker_error(error)
           build.report_worker_error(error)
         end
 
-        def acknowledge(test_key)
+        def acknowledge(test_key, error: nil, pipeline: redis)
           raise_on_mismatching_test(test_key)
           eval_script(
             :acknowledge,
-            keys: [key('running'), key('processed'), key('owners')],
-            argv: [test_key],
+            keys: [key('running'), key('processed'), key('owners'), key('error-reports')],
+            argv: [test_key, error.to_s, config.redis_ttl],
+            pipeline: pipeline,
           ) == 1
         end
 

ruby/lib/ci/queue/static.rb

@@ -103,12 +103,12 @@ def exhausted?
         @queue.empty?
       end
 
-      def acknowledge(test)
+      def acknowledge(*)
         @progress += 1
         true
       end
 
-      def increment_test_failed
+      def increment_test_failed(*)
         @test_failed = test_failed + 1
       end