Only increment stats on ack; delta-based stats; stat correction on replace

- Record stats only when worker acknowledges; duplicate acks do not increment
- Redis: record_stats_delta (HINCRBY); record_success returns true when ack'd or replaced
- Stat correction when success replaces failure; real assertion count (test.assertions) in delta
- Test helper: Requeue before Skip when both set; test_aggregation and integration expectations updated
- Remove [stats] debug logging from Redis BuildRecord; test_redis_reporter assertions = 8

Author: kangze-jia

ruby/lib/ci/queue/build_record.rb

@@ -18,18 +18,35 @@ def queue_exhausted?
         @queue.exhausted?
       end
 
-      def record_error(id, payload, stats: nil)
+      def record_error(id, payload, stat_delta: nil)
         error_reports[id] = payload
-        record_stats(stats)
+        true
       end
 
-      def record_success(id, stats: nil, skip_flaky_record: false, acknowledge: true)
+      def record_success(id, skip_flaky_record: false, acknowledge: true)
         error_reports.delete(id)
-        record_stats(stats)
+        true
+      end
+
+      def record_requeue(id)
+        true
+      end
+
+      def record_stats(builds_stats)
+        return unless builds_stats
+        stats.merge!(builds_stats)
+      end
+
+      def record_stats_delta(delta, pipeline: nil)
+        return if delta.nil? || delta.empty?
+        delta.each do |stat_name, value|
+          next unless value.is_a?(Numeric) || value.to_s.match?(/\A-?\d+\.?\d*\z/)
+          stats[stat_name] = (stats[stat_name] || 0).to_f + value.to_f
+        end
       end
 
       def fetch_stats(stat_names)
-        stat_names.zip(stats.values_at(*stat_names).map(&:to_f))
+        stat_names.zip(stats.values_at(*stat_names).map(&:to_f)).to_h
       end
 
       def reset_stats(stat_names)
@@ -47,11 +64,6 @@ def worker_errors
       private
 
       attr_reader :stats
-
-      def record_stats(builds_stats)
-        return unless builds_stats
-        stats.merge!(builds_stats)
-      end
     end
   end
 end

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

@@ -56,31 +56,71 @@ def record_warning(type, attributes)
           redis.rpush(key('warnings'), Marshal.dump([type, attributes]))
         end
 
-        def record_error(id, payload, stats: nil)
+        def record_error(id, payload, stat_delta: nil)
+          # Run acknowledge first so we know whether we're the first to ack
           acknowledged = @queue.acknowledge(id, error: payload)
 
           if acknowledged
-            # if another worker already acknowledged the test, we don't need to update the global stats or increment the test failed count
-            record_stats(stats)
+            # We were the first to ack; another worker already ack'd would get falsy from SADD
             @queue.increment_test_failed
+            # Only the acknowledging worker's stats include this failure (others skip increment when ack=false).
+            # Store so we can subtract it if another worker records success later.
+            store_error_report_delta(id, stat_delta) if stat_delta && stat_delta.any?
           end
-          nil
+          # Return so caller can roll back local counter when not acknowledged
+          !!acknowledged
         end
 
-        def record_success(id, stats: nil, skip_flaky_record: false)
-          _, error_reports_deleted_count, requeued_count, _ = redis.multi do |transaction|
+        def record_success(id, skip_flaky_record: false)
+          acknowledged, error_reports_deleted_count, requeued_count, delta_json = redis.multi do |transaction|
             @queue.acknowledge(id, pipeline: transaction)
             transaction.hdel(key('error-reports'), id)
             transaction.hget(key('requeues-count'), id)
-            record_stats(stats, pipeline: transaction)
+            transaction.hget(key('error-report-deltas'), id)
+          end
+          # When we're replacing a failure, subtract the (single) acknowledging worker's stat contribution
+          if error_reports_deleted_count.to_i > 0 && delta_json
+            apply_error_report_delta_correction(delta_json)
+            redis.hdel(key('error-report-deltas'), id)
           end
           record_flaky(id) if !skip_flaky_record && (error_reports_deleted_count.to_i > 0 || requeued_count.to_i > 0)
-          nil
+          # Count this run when we ack'd or when we replaced a failure (so stats delta is applied)
+          !!(acknowledged || error_reports_deleted_count.to_i > 0)
         end
 
-        def record_requeue(id, stats: nil)
-          redis.pipelined do |pipeline|
-           record_stats(stats, pipeline: pipeline)
+        def record_requeue(id)
+          true
+        end
+
+        def record_stats(stats = nil, pipeline: nil)
+          return unless stats
+          if pipeline
+            stats.each do |stat_name, stat_value|
+              pipeline.hset(key(stat_name), config.worker_id, stat_value)
+              pipeline.expire(key(stat_name), config.redis_ttl)
+            end
+          else
+            redis.pipelined do |p|
+              record_stats(stats, pipeline: p)
+            end
+          end
+        end
+
+        # Apply a delta to this worker's stats in Redis (HINCRBY). Use this instead of
+        # record_stats when recording per-test so we never overwrite and correction sticks.
+        def record_stats_delta(delta, pipeline: nil)
+          return if delta.nil? || delta.empty?
+          apply_delta = lambda do |p|
+            delta.each do |stat_name, value|
+              next unless value.is_a?(Numeric) || value.to_s.match?(/\A-?\d+\.?\d*\z/)
+              p.hincrbyfloat(key(stat_name), config.worker_id.to_s, value.to_f)
+              p.expire(key(stat_name), config.redis_ttl)
+            end
+          end
+          if pipeline
+            apply_delta.call(pipeline)
+          else
+            redis.pipelined { |p| apply_delta.call(p) }
           end
         end
 
@@ -131,17 +171,31 @@ def reset_stats(stat_names)
 
         attr_reader :config, :redis
 
-        def record_stats(stats, pipeline: redis)
-          return unless stats
-          stats.each do |stat_name, stat_value|
-            pipeline.hset(key(stat_name), config.worker_id, stat_value)
-            pipeline.expire(key(stat_name), config.redis_ttl)
-          end
-        end
-
         def key(*args)
           KeyShortener.key(config.build_id, *args)
         end
+
+        def store_error_report_delta(test_id, stat_delta)
+          # Only the acknowledging worker's stats include this test; store their delta for correction on success
+          payload = { 'worker_id' => config.worker_id.to_s }.merge(stat_delta)
+          redis.hset(key('error-report-deltas'), test_id, JSON.generate(payload))
+          redis.expire(key('error-report-deltas'), config.redis_ttl)
+        end
+
+        def apply_error_report_delta_correction(delta_json)
+          delta = JSON.parse(delta_json)
+          worker_id = delta.delete('worker_id')&.to_s
+          return if worker_id.nil? || worker_id.empty? || delta.empty?
+
+          redis.pipelined do |pipeline|
+            delta.each do |stat_name, value|
+              next unless value.is_a?(Numeric) || value.to_s.match?(/\A-?\d+\.?\d*\z/)
+
+              pipeline.hincrbyfloat(key(stat_name), worker_id, -value.to_f)
+              pipeline.expire(key(stat_name), config.redis_ttl)
+            end
+          end
+        end
       end
     end
   end

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

@@ -10,20 +10,32 @@ def initialize(queue, redis, config)
           @config = config
         end
 
-        def record_error(payload, stats: nil)
+        def record_error(payload)
           redis.pipelined do |pipeline|
             pipeline.lpush(
               key('error-reports'),
               payload,
             )
             pipeline.expire(key('error-reports'), config.redis_ttl)
-            record_stats(stats, pipeline: pipeline)
           end
           nil
         end
 
-        def record_success(stats: nil)
-          record_stats(stats)
+        def record_success
+        end
+
+        def record_stats(stats, pipeline: nil)
+          return unless stats
+          if pipeline
+            stats.each do |stat_name, stat_value|
+              pipeline.hset(key(stat_name), config.worker_id, stat_value)
+              pipeline.expire(key(stat_name), config.redis_ttl)
+            end
+          else
+            redis.pipelined do |p|
+              record_stats(stats, pipeline: p)
+            end
+          end
         end
 
         def record_warning(_,_)
@@ -54,14 +66,6 @@ def pop_warnings
         def key(*args)
           KeyShortener.key(config.build_id, *args)
         end
-
-        def record_stats(stats, pipeline: redis)
-          return unless stats
-          stats.each do |stat_name, stat_value|
-            pipeline.hset(key(stat_name), config.worker_id, stat_value)
-            pipeline.expire(key(stat_name), config.redis_ttl)
-          end
-        end
       end
     end
   end

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

@@ -12,6 +12,7 @@ module KeyShortener
           'queue' => 'q',
           'owners' => 'o',
           'error-reports' => 'e',
+          'error-report-deltas' => 'ed',
           'requeues-count' => 'rc',
           'assertions' => 'a',
           'errors' => 'er',

ruby/lib/minitest/queue/build_status_recorder.rb

@@ -38,28 +38,46 @@ def record(test)
         super
 
         self.total_time = Minitest.clock_time - start_time
-        if test.requeued?
-          self.requeues += 1
-        elsif test.skipped?
-          self.skips += 1
-        elsif test.error?
-          self.errors += 1
-        elsif test.failure
-          self.failures += 1
-        end
+        
+        # Determine what type of result this is and record it
+        test_id = "#{test.klass}##{test.name}"
+        delta = delta_for(test)
 
-        stats = COUNTERS.zip(COUNTERS.map { |c| send(c) }).to_h
-        if (test.failure || test.error?) && !test.skipped?
-          build.record_error("#{test.klass}##{test.name}", dump(test), stats: stats)
+        acknowledged = if (test.failure || test.error?) && !test.skipped?
+          build.record_error(test_id, dump(test), stat_delta: delta)
         elsif test.requeued?
-          build.record_requeue("#{test.klass}##{test.name}", stats: stats)
+          build.record_requeue(test_id)
         else
-          build.record_success("#{test.klass}##{test.name}", stats: stats, skip_flaky_record: test.skipped?)
+          build.record_success(test_id, skip_flaky_record: test.skipped?)
+        end
+
+        if acknowledged
+          if (test.failure || test.error?) && !test.skipped?
+            test.error? ? self.errors += 1 : self.failures += 1
+          elsif test.requeued?
+            self.requeues += 1
+          elsif test.skipped?
+            self.skips += 1
+          end
+          # Apply delta to Redis (record_success returns true when ack'd or when we replaced a failure)
+          build.record_stats_delta(delta)
         end
       end
 
       private
 
+      def delta_for(test)
+        h = { 'assertions' => (test.assertions || 0).to_i, 'errors' => 0, 'failures' => 0, 'skips' => 0, 'requeues' => 0, 'total_time' => test.time.to_f }
+        if (test.failure || test.error?) && !test.skipped?
+          test.error? ? h['errors'] = 1 : h['failures'] = 1
+        elsif test.requeued?
+          h['requeues'] = 1
+        elsif test.skipped?
+          h['skips'] = 1
+        end
+        h
+      end
+
       def dump(test)
         ErrorReport.new(self.class.failure_formatter.new(test).to_h).dump
       end

ruby/lib/minitest/queue/grind_recorder.rb

@@ -32,12 +32,12 @@ def record(test)
       private
 
       def record_test(test)
-        stats = self.class.counters
         if (test.failure || test.error?) && !test.skipped?
-          build.record_error(dump(test), stats: stats)
+          build.record_error(dump(test))
         else
-          build.record_success(stats: stats)
+          build.record_success
         end
+        build.record_stats(self.class.counters)
       end
 
       def increment_counter(test)

ruby/test/integration/minitest_redis_test.rb

@@ -88,6 +88,7 @@ def test_lost_test_with_heartbeat_monitor
 
         assert_empty err
         result = normalize(out.lines[1].strip)
+        # lost_test.rb test_foo has no assertions (only sleep)
         assert_equal "Ran 1 tests, 0 assertions, 0 failures, 0 errors, 0 skips, 0 requeues in X.XXs (aggregated)", result
         warnings = warnings_file.read.lines.map { |line| JSON.parse(line) }
         assert_equal 1, warnings.size
@@ -513,7 +514,8 @@ def test_retry_report
 
       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: {
+        queue.build.record_success(test_id.dup)
+        queue.build.record_stats({
           'assertions' => index + 1,
           'errors' => 0,
           'failures' => 0,
@@ -912,6 +914,7 @@ def test_redis_reporter
       end
       assert_empty err
       output = normalize(out.lines.last.strip)
+      # 8 = sum of test.assertions from Minitest (skip counts as 1 in some versions)
       assert_equal 'Ran 11 tests, 8 assertions, 2 failures, 1 errors, 1 skips, 4 requeues in X.XXs', output
 
       Tempfile.open('warnings') do |warnings_file|