Class: JobBoard::ProcessTree

Inherits:
Object
  • Object
show all
Defined in:
app/models/job_board/process_tree.rb

Overview

Supervisor -> supervisee tree of SolidQueue processes with staleness flags, claimed-job counts, and the in-progress jobs each worker holds.

Defined Under Namespace

Classes: Node

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(stale_threshold:) ⇒ ProcessTree

Returns a new instance of ProcessTree.



13
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
 
# File 'app/models/job_board/process_tree.rb', line 13

def initialize(stale_threshold:)
  processes = SolidQueue::Process.order(:id).to_a
  ids = processes.map(&:id).to_set
  counts = SolidQueue::ClaimedExecution.group(:process_id).count
  @claimed_by_process =
    SolidQueue::ClaimedExecution.where(process_id: processes.map(&:id))
      .includes(:job).group_by(&:process_id)

  cutoff = stale_threshold.ago
  children = processes.group_by(&:supervisor_id)
  build_node = nil
  build_node = lambda do |process|
    Node.new(
      process: process,
      children: (children[process.id] || []).map(&build_node),
      claimed_count: counts[process.id] || 0,
      stale: process.last_heartbeat_at < cutoff
    )
  end

  # Roots: no supervisor, or a supervisor that no longer exists.
  root_processes = processes.select { |p| p.supervisor_id.nil? || !ids.include?(p.supervisor_id) }
  supervisors, others = root_processes.partition { |p| p.kind.to_s.start_with?("Supervisor") }
  @roots = (supervisors + others).map(&build_node)

  @orphaned_claimed_count = SolidQueue::ClaimedExecution.orphaned.count
end

Instance Attribute Details

#orphaned_claimed_count ⇒ Object (readonly)

Returns the value of attribute orphaned_claimed_count.



7
8
9
 
# File 'app/models/job_board/process_tree.rb', line 7

def orphaned_claimed_count
  @orphaned_claimed_count
end

#roots ⇒ Object (readonly)

Returns the value of attribute roots.



7
8
9
 
# File 'app/models/job_board/process_tree.rb', line 7

def roots
  @roots
end

Class Method Details

.build(stale_threshold:) ⇒ Object 



9
10
11
 
# File 'app/models/job_board/process_tree.rb', line 9

def self.build(stale_threshold:)
  new(stale_threshold: stale_threshold)
end

Instance Method Details

#claimed_executions_for(process) ⇒ Object 



41
42
43
 
# File 'app/models/job_board/process_tree.rb', line 41

def claimed_executions_for(process)
  @claimed_by_process[process.id] || []
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


45
46
47
 
# File 'app/models/job_board/process_tree.rb', line 45

def empty?
  roots.empty?
end