Class: ScatterGather::Completion

Inherits:
ActiveRecord::Base
  • Object
show all
Defined in:
lib/scatter_gather/completion.rb

Overview

ActiveJob itself does not maintain a "one record per job" database entry where we could reliably track job completion. The underlying job storage (e.g., Sidekiq, Que, DelayedJob) is determined by the ActiveJob adapter, and their implementation details vary. As a result, to enable reliable scatter-gather orchestration, we create our own completion tracking records. Each job that is meant to be gathered later inserts a record into the scatter_gather_completions table when it is enqueued, and marks it as completed upon finishing. This ensures adapter-independent, uniform dependency tracking.

Class Method Summary collapse

Class Method Details

.all_dependencies_completed?(dependency_statuses) ⇒ Boolean

Check if all dependencies are completed

Parameters:

  • dependency_statuses (Array<DependencyStatus>) —

    Array of dependency statuses

Returns:

  • (Boolean) —

    true if all dependencies are completed



54
55
56
 
# File 'lib/scatter_gather/completion.rb', line 54

def self.all_dependencies_completed?(dependency_statuses)
  dependency_statuses.all? { |ds| ds.status == :completed }
end

.collect_statuses(active_job_ids) ⇒ Array<DependencyStatus>

Collect status information for the given active job IDs

Parameters:

  • active_job_ids (Array<String>) —

    Array of ActiveJob IDs to check

Returns:



14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/scatter_gather/completion.rb', line 14

def self.collect_statuses(active_job_ids)
  # Initialize all job IDs with unknown status
  statuses = active_job_ids.map { |id| [id, :unknown] }.to_h

  # Get statuses from completion records
  completions = where(active_job_id: active_job_ids)
    .pluck(:active_job_id, :active_job_class_name, :status)
    .map do |(id, class_name, status)|
      [id, {class_name: class_name, status: status.to_sym}]
    end.to_h

  # Update statuses with completion data
  completions.each do |id, data|
    statuses[id] = data[:status]
  end

  # Create DependencyStatus objects
  dependency_statuses = active_job_ids.map do |id|
    completion_data = completions[id]
    class_name = completion_data&.dig(:class_name)
    status = statuses[id]

    ScatterGather::DependencyStatus.new(id, class_name, status)
  end

  # Sort by status first (unknown, pending, completed), then by active_job_id
  dependency_statuses.sort_by do |ds|
    status_order = case ds.status
    when :unknown then 0
    when :pending then 1
    when :completed then 2
    else 3
    end
    [status_order, ds.active_job_id.to_s]
  end
end