Class: Minitest::Distributed::Coordinators::RedisCoordinator

Inherits:
Object
  • Object
show all
Extended by:
T::Sig
Includes:
CoordinatorInterface
Defined in:
lib/minitest/distributed/coordinators/redis_coordinator.rb

Overview

The RedisCoordinator is an implementation of the test coordinator interface using a Redis stream + consumergroup for coordination.

We assume a bunch of workers will be started at the same time. Every worker will try to become the leader by trying to create the consumergroup. Only one will succeed, which will then continue to populate the list of tests to run to the stream.

AFter that, all workers will start consuming from the stream. They will first try to claim stale entries from other workers (determined by the test_timeout_seconds option), and process them up to a maximum of max_attempts attempts. Then, they will consume tests from the stream, run them, and ack them. This is done in batches to reduce load on Redis.

Retrying failed tests (up to max_attempts times) uses the same mechanism. When a test fails, and we haven't exhausted the maximum number of attempts, we do not ACK the result with Redis. The means that another worker will eventually claim the test, and run it again. However, in this case we don't want to slow things down unnecessarily. When a test fails and we want to retry it, we add the test to the retry_set in Redis. When other worker sees that a test is in this set, it can immediately claim the test, rather than waiting for the timeout.

Finally, when we have acked the same number of tests as we populated into the queue, the run is considered finished. The first worker to detect this will remove the consumergroup and the associated stream from Redis.

If a worker starts for the same run_id while it is already considered completed, it will start a "retry run". It will find all the tests that failed/errored on the previous attempt, and schedule only those tests to be run, rather than the full test suite returned by the test selector. This can be useful to retry flaky tests. Subsequent workers coming online will join this worker to form a consumer group exactly as described above.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(configuration:) ⇒ RedisCoordinator

Returns a new instance of RedisCoordinator.



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 90

def initialize(configuration:)
  @configuration = configuration

  @redis = T.let(nil, T.nilable(Redis))
  @register_consumergroup_script = T.let(nil, T.nilable(String))
  @read_results_script = T.let(nil, T.nilable(String))
  @reset_results_script = T.let(nil, T.nilable(String))
  @commit_results_script = T.let(nil, T.nilable(String))
  @cleanup_script = T.let(nil, T.nilable(String))
  @abort_script = T.let(nil, T.nilable(String))
  @adjust_results_script = T.let(nil, T.nilable(String))
  @publish_tests_script = T.let(nil, T.nilable(String))
  @heartbeat_script = T.let(nil, T.nilable(String))
  @mark_attempt_state_script = T.let(nil, T.nilable(String))
  @stream_key = T.let(key("queue"), String)
  @group_name = T.let(BASE_GROUP_NAME, String)
  @attempt_generation = T.let(nil, T.nilable(String))
  @production_heartbeat_error = T.let(nil, T.nilable(StandardError))
  @production_heartbeat_control = T.let(nil, T.nilable(HeartbeatControl))
  @mutating_script_mutex = T.let(Mutex.new, Mutex)
  @local_results = T.let(ResultAggregate.new, ResultAggregate)
  @combined_results = T.let(nil, T.nilable(ResultAggregate))
  @combined_results_production_complete = T.let(nil, T.nilable(T::Boolean))
  @attempt_truncated = T.let(false, T::Boolean)
  @truncated_follower = T.let(false, T::Boolean)
  @retry_refused_due_to_truncation = T.let(false, T::Boolean)
  @truncation_state_invalid = T.let(false, T::Boolean)
  @registration_rejected = T.let(false, T::Boolean)
  @superseded = T.let(false, T::Boolean)
  @reclaimed_timeout_tests = T.let(Set.new, T::Set[EnqueuedRunnable])
  @reclaimed_failed_tests = T.let(Set.new, T::Set[EnqueuedRunnable])
  @aborted = T.let(false, T::Boolean)
  @stall_diagnostic = T.let(nil, T.nilable(String))
  @output = T.let(nil, T.untyped)
end

Instance Attribute Details

#configuration ⇒ Object (readonly)

Returns the value of attribute configuration.



69
70
71
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 69

def configuration
  @configuration
end

#group_name ⇒ Object (readonly)

Returns the value of attribute group_name.



75
76
77
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 75

def group_name
  @group_name
end

#local_results ⇒ Object (readonly)

Returns the value of attribute local_results.



78
79
80
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 78

def local_results
  @local_results
end

#reclaimed_failed_tests ⇒ Object (readonly)

Returns the value of attribute reclaimed_failed_tests.



84
85
86
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 84

def reclaimed_failed_tests
  @reclaimed_failed_tests
end

#reclaimed_timeout_tests ⇒ Object (readonly)

Returns the value of attribute reclaimed_timeout_tests.



81
82
83
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 81

def reclaimed_timeout_tests
  @reclaimed_timeout_tests
end

#stall_diagnostic ⇒ Object (readonly)

Returns the value of attribute stall_diagnostic.



87
88
89
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 87

def stall_diagnostic
  @stall_diagnostic
end

#stream_key ⇒ Object (readonly)

Returns the value of attribute stream_key.



72
73
74
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 72

def stream_key
  @stream_key
end

Instance Method Details

#aborted? ⇒ Boolean

Returns:

  • (Boolean) 


167
168
169
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 167

def aborted?
  @aborted
end

#combined_results ⇒ Object 



133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 133

def combined_results
  @combined_results ||= begin
    keys = STATS_KEY_NAMES.map { |name| key(name) }
    keys << key("production_complete")
    response = T.cast(
      execute_script(script_name: :read_results, keys: keys, argv: []),
      T::Array[T.untyped],
    )
    @combined_results_production_complete = response.fetch(0) == 1
    stats_as_string = response.drop(1)

    ResultAggregate.new(
      max_failures: configuration.max_failures,

      runs: Integer(stats_as_string.fetch(0).then { |value| value == "" ? 0 : value }),
      assertions: Integer(stats_as_string.fetch(1).then { |value| value == "" ? 0 : value }),
      passes: Integer(stats_as_string.fetch(2).then { |value| value == "" ? 0 : value }),
      failures: Integer(stats_as_string.fetch(3).then { |value| value == "" ? 0 : value }),
      errors: Integer(stats_as_string.fetch(4).then { |value| value == "" ? 0 : value }),
      skips: Integer(stats_as_string.fetch(5).then { |value| value == "" ? 0 : value }),
      requeues: Integer(stats_as_string.fetch(6).then { |value| value == "" ? 0 : value }),
      discards: Integer(stats_as_string.fetch(7).then { |value| value == "" ? 0 : value }),
      acks: Integer(stats_as_string.fetch(8).then { |value| value == "" ? 0 : value }),

      # Before the producer initializes counters, a sentinel prevents the
      # absent size from making an unpublished run appear complete.
      size: Integer(stats_as_string.fetch(9).then { |value| value == "" ? 2_147_483_647 : value }),
    )
  end
rescue ArgumentError, TypeError => parse_error
  raise CoordinatorStateError, "invalid Redis aggregate: #{parse_error.message}"
end

#consume(reporter:) ⇒ Object 



380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 380

def consume(reporter:)
  return if @retry_refused_due_to_truncation || @registration_rejected || @truncation_state_invalid || superseded?

  exponential_backoff = INITIAL_BACKOFF
  last_progress_at = monotonic_time
  initial_results = combined_results
  observed_acks = T.let(initial_results.acks, T.nilable(Integer))
  observed_size = T.let(initial_results.size, T.nilable(Integer))
  observed_production_heartbeat = current_production_heartbeat
  pending_stall_warning_emitted = T.let(false, T::Boolean)
  drained_mismatch_detected_at = T.let(nil, T.nilable(Float))
  incomplete_production_detected_at = T.let(nil, T.nilable(Float))

  loop do
    # Other workers can advance or complete the run without touching this
    # process's memoized aggregate. Refresh once per polling iteration.
    @combined_results = nil

    # First, see if there are any pending tests from other workers to claim.
    stale_runnables = claim_stale_runnables
    process_batch(stale_runnables, reporter)
    break if commit_observed_truncation? || superseded?

    # Then, try to process a regular batch of messages
    fresh_runnables = claim_fresh_runnables(block: exponential_backoff)
    process_batch(fresh_runnables, reporter)
    break if commit_observed_truncation? || superseded?

    run_results = combined_results

    # If we have acked the same amount of tests as we were supposed to, the run
    # is complete and we can exit our loop. Generally, only one worker will detect
    # this condition. The other workers will quit their consumer loop because the
    # consumer group will be deleted by the first worker, and their Redis commands
    # will start to fail - see the rescue block below. Counters can be 0/0 before
    # an empty run finishes publishing, so production completion is also required.
    break if run_complete?(run_results)

    # We also abort a run if we reach the maximum number of failures.
    if run_results.abort? && !run_results.complete?
      mark_run_truncated
      break
    end

    processed_batch = stale_runnables.any? || fresh_runnables.any?
    if processed_batch
      last_progress_at = monotonic_time
      observed_acks = run_results.acks
      observed_size = run_results.size
      drained_mismatch_detected_at = nil
      incomplete_production_detected_at = nil
    else
      now = monotonic_time
      confirmation_interval = [configuration.stall_timeout_seconds, STALL_CONFIRMATION_SECONDS].min
      confirmation_due = if drained_mismatch_detected_at
        now - drained_mismatch_detected_at >= confirmation_interval
      elsif incomplete_production_detected_at
        now - incomplete_production_detected_at >= configuration.stall_timeout_seconds
      else
        true
      end

      if now - last_progress_at >= configuration.stall_timeout_seconds && confirmation_due
        probe = probe_stall
        @combined_results = nil

        if probe.invalid_values.any?
          abort_with_diagnostic(<<~DIAGNOSTIC)
            ERROR: minitest-distributed found invalid numeric Redis coordinator state.
            #{format_probe_state(probe)}
            Invalid values: #{probe.invalid_values.join(", ")}
          DIAGNOSTIC
          break
        end

        # Another worker may have completed the run while our memoized aggregate
        # was stale. Treat the fresh counters as authoritative.
        if probe.production_complete && !probe.acks.nil? && probe.acks == probe.size
          # The probe checks completion/liveness fields only. Validate all
          # retained statistics and the same-snapshot production marker
          # before accepting terminal success.
          terminal_results = combined_results
          break if run_complete?(terminal_results)
        end

        counters_changed = probe.acks != observed_acks || probe.size != observed_size
        heartbeat_changed = probe.production_heartbeat != observed_production_heartbeat
        if counters_changed
          observed_acks = probe.acks
          observed_size = probe.size
        end
        observed_production_heartbeat = probe.production_heartbeat

        stream_empty = probe.pending_count.zero? && probe.lag == 0
        if probe.production_complete && stream_empty
          incomplete_production_detected_at = nil
          if drained_mismatch_detected_at && !counters_changed
            abort_stalled_run(probe)
            break
          else
            # The probe is pipelined rather than transactional. Confirm an
            # unchanged mismatch so a final atomic commit interleaved with the
            # probe cannot cause a false abort.
            drained_mismatch_detected_at = now
          end
        elsif !probe.production_complete && stream_empty
          drained_mismatch_detected_at = nil
          production_progressed = counters_changed || heartbeat_changed
          if incomplete_production_detected_at && !production_progressed
            abort_stalled_run(probe)
            break
          elsif production_progressed
            # A live leader refreshes this heartbeat while it selects and
            # publishes tests, so arbitrarily slow discovery is not mistaken
            # for a dead producer.
            incomplete_production_detected_at = nil
            last_progress_at = now
          else
            # No producer heartbeat was observed. Confirm once more after a
            # full stall interval before declaring the leader dead.
            incomplete_production_detected_at = now
          end
        else
          drained_mismatch_detected_at = nil
          incomplete_production_detected_at = nil

          if probe.pending_count > 0 && !pending_stall_warning_emitted
            emit_message(format_pending_stall_warning(probe))
            pending_stall_warning_emitted = true
          end

          # A non-empty PEL may be waiting for the legitimate
          # test_timeout_seconds * test_batch_size reclaim path. Undelivered
          # entries can likewise be claimed by another worker. Wait another
          # full interval before probing again.
          last_progress_at = now
        end
      end
    end

    # To make sure we don't end up in a busy loop overwhelming Redis with commands
    # when there is no work to do, we increase the blocking time exponentially,
    # and reset it to the initial value if we processed any tests.
    #
    # The backoff is capped at MAX_BACKOFF to bound how long a worker can sit
    # inside a single XREADGROUP BLOCK call. Without a cap, after ~15 empty
    # iterations the worker is blocked in Redis for 5+ minutes and cannot
    # re-check `complete?` / `abort?` until the BLOCK returns, which manifests
    # as a long post-100% teardown hang when pipelined XACKs race the progress
    # reporter.
    exponential_backoff = if processed_batch
      INITIAL_BACKOFF
    else
      next_backoff(exponential_backoff)
    end
  end

  cleanup
rescue Redis::CommandError => ce
  coordinator_state_error = ce.message.start_with?(
    "NOGROUP",
    "WRONGTYPE",
    "COORDINATORSTATE",
    "COORDINATORSTREAM",
    "STALEATTEMPT",
    "COORDINATORCONFIG",
  ) || ce.message.include?("no such key")
  if coordinator_state_error
    # A normal cleanup and missing/evicted state can produce similar Redis
    # errors. Fresh counters distinguish a terminal run from data loss.
    handle_coordinator_state_error(ce)
    cleanup if stalled?
  else
    raise
  end
rescue CoordinatorStateError => parse_error
  abort_with_diagnostic(<<~DIAGNOSTIC)
    ERROR: minitest-distributed could not parse Redis coordinator state.
    run_id=#{configuration.run_id} worker_id=#{configuration.worker_id}
    parse_error=#{parse_error.message.inspect}
  DIAGNOSTIC
  cleanup if stalled?
ensure
  # Another worker may commit the final batch and clean up while this
  # worker is unwinding from NOGROUP. Report and validate against one
  # last fresh aggregate rather than a pre-cleanup local cache.
  @combined_results = nil
end

#current_attempt_truncated? ⇒ Boolean

Returns:

  • (Boolean) 


210
211
212
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 210

def current_attempt_truncated?
  @attempt_truncated
end

#persisted_truncation? ⇒ Boolean

Returns:

  • (Boolean) 


183
184
185
186
187
188
189
190
191
192
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 183

def persisted_truncation?
  truncated_key = key("truncated")
  generation_key = key("truncated_generation")
  return false unless redis.type(truncated_key) == "string"
  return false unless redis.type(generation_key) == "string"

  redis.get(truncated_key) == "1" && !redis.get(generation_key).to_s.empty?
rescue Redis::BaseError
  false
end

#produce(test_selector:) ⇒ Object 



225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 225

def produce(test_selector:)
  production_heartbeat_thread = T.let(nil, T.nilable(Thread))
  propagate_heartbeat_error = T.let(false, T::Boolean)

  registration_keys = [
    stream_key,
    key("attempt_generation"),
    key("stalled"),
    key("production_complete"),
    key("production_heartbeat"),
    key("retry_set"),
  ]
  registration_keys.concat(STATS_KEY_NAMES.map { |name| key(name) })
  registration_keys.concat(LIST_KEY_RESULT_TYPES.map { |result_type| list_key(result_type.serialize) })
  registration_keys.push(
    key("truncated"),
    key("completed_at"),
    key("retry_snapshot_digest"),
    key("truncated_generation"),
    key("retention_ttl"),
  )
  registration = T.let(nil, T.untyped)
  registration_mode = T.let(-1, Integer)
  leader = T.let(false, T::Boolean)
  awaited_generation = T.let(nil, T.nilable(String))
  loop do
    registration = T.unsafe(execute_script(
      script_name: :register_consumergroup,
      keys: registration_keys,
      argv: [
        BASE_GROUP_NAME,
        configuration.key_ttl_seconds,
        configuration.max_failures || "",
        SecureRandom.uuid,
        configuration.completion_grace_seconds,
        awaited_generation || "",
      ],
    ))

    leader = registration.fetch(0) == 1
    @attempt_generation = String(registration.fetch(1))
    @group_name = "#{BASE_GROUP_NAME}-#{@attempt_generation}"
    registration_mode = Integer(registration.fetch(2))
    if registration_mode == -3
      @aborted = true
      @truncated_follower = true
      @retry_refused_due_to_truncation = true
    elsif registration_mode == -4
      @aborted = true
      @registration_rejected = true
      emit_message(<<~ERROR)
        ERROR: minitest-distributed rejected a Redis key TTL change for an existing run.
        run_id=#{configuration.run_id} worker_id=#{configuration.worker_id}
        Active workers must use the same key TTL, and later retries cannot decrease the retained TTL.
      ERROR
    elsif registration_mode == -5
      @aborted = true
      @truncation_state_invalid = true
      emit_message(<<~ERROR)
        ERROR: minitest-distributed cannot retry because the retained truncation fence is incomplete.
        run_id=#{configuration.run_id} worker_id=#{configuration.worker_id}
        The run remains failed; restore or expire its retained state before reusing this run ID.
      ERROR
    end
    break unless registration_mode == -2

    sleep(Float(registration.fetch(5)))
    awaited_generation = String(registration.fetch(1))
  end
  return unless leader

  previous_failures = T.cast(registration.fetch(3), T::Array[String])
  previous_errors = T.cast(registration.fetch(4), T::Array[String])
  production_heartbeat_thread = start_production_heartbeat unless registration_mode == 3

  tests = T.let(
    case registration_mode
    when 0 # First attempt for a new run ID.
      tests_from_selector = test_selector.tests
      adjust_combined_results(
        ResultAggregate.new(size: tests_from_selector.size),
        allow_missing_stats: true,
      )
      tests_from_selector
    when 1 # Valid completed attempt; use selective retry behavior.
      if configuration.retry_failures
        test_identifiers_to_retry = T.let(previous_failures + previous_errors, T::Array[String])
        retry_tests = materialize_retry_tests(test_identifiers_to_retry)
        if retry_tests
          total_failures = retry_tests.length
          adjust_combined_results(
            ResultAggregate.new(
              size: total_failures,
              failures: -previous_failures.length,
              errors: -previous_errors.length,
              requeues: total_failures,
            ),
            clear_retry_lists: true,
          )
          retry_tests
        else
          emit_message(<<~WARNING)
            WARNING: The previous attempt retained an invalid retry identifier.
            Running the full test suite instead of a selective retry.
          WARNING
          tests_from_selector = test_selector.tests
          reset_combined_results_for_full_rerun(size: tests_from_selector.size)
          tests_from_selector
        end
      else
        adjust_combined_results(ResultAggregate.new(size: 0))
        []
      end
    when 2 # Inconsistent or explicitly stalled state; rerun everything.
      emit_message(<<~WARNING)
        WARNING: The previous attempt lost Redis coordinator state.
        Running the full test suite instead of a selective retry.
      WARNING
      tests_from_selector = test_selector.tests
      adjust_combined_results(
        ResultAggregate.new(size: tests_from_selector.size),
        allow_missing_stats: true,
      )
      tests_from_selector
    when 3 # The previous attempt intentionally stopped at max_failures.
      @aborted = true
      @retry_refused_due_to_truncation = true
      []
    else
      raise "Unknown Redis registration mode: #{registration_mode}"
    end,
    T::Array[Minitest::Runnable],
  )

  unless @retry_refused_due_to_truncation
    begin
      publish_tests(tests)
      propagate_heartbeat_error = true
    rescue Redis::CommandError => error
      raise unless handle_publish_race(error)
    end
  end
ensure
  if production_heartbeat_thread
    stop_production_heartbeat(
      production_heartbeat_thread,
      propagate_error: propagate_heartbeat_error,
    )
  end
end

#register_reporters(reporter:, options:) ⇒ Object 



127
128
129
130
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 127

def register_reporters(reporter:, options:)
  @output = options[:io]
  reporter << Reporters::RedisCoordinatorWarningsReporter.new(options[:io], options)
end

#registration_rejected? ⇒ Boolean

Returns:

  • (Boolean) 


200
201
202
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 200

def registration_rejected?
  @registration_rejected
end

#retry_refused_due_to_truncation? ⇒ Boolean

Returns:

  • (Boolean) 


205
206
207
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 205

def retry_refused_due_to_truncation?
  @retry_refused_due_to_truncation
end

#stalled? ⇒ Boolean

Returns:

  • (Boolean) 


195
196
197
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 195

def stalled?
  !stall_diagnostic.nil?
end

#superseded? ⇒ Boolean

Returns:

  • (Boolean) 


220
221
222
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 220

def superseded?
  @superseded
end

#truncation_state_invalid? ⇒ Boolean

Returns:

  • (Boolean) 


215
216
217
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 215

def truncation_state_invalid?
  @truncation_state_invalid
end

#valid_combined_results? ⇒ Boolean

Returns:

  • (Boolean) 


172
173
174
175
176
177
178
179
180
 
# File 'lib/minitest/distributed/coordinators/redis_coordinator.rb', line 172

def valid_combined_results?
  # Reporter success must be based on one fresh atomic snapshot. Counters
  # alone can look complete after production_complete is evicted.
  @combined_results = nil
  results = combined_results
  results.valid? && @combined_results_production_complete == true
rescue Redis::BaseError, CoordinatorStateError
  false
end