Class: Wurk::Flow::Status

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/flow/status.rb

Overview

A created flow, read back. The counterpart of Batch::Status, and the only reader of the flow key schema: the dashboard controller and the machine API both go through this, so the two surfaces cannot disagree about what state a flow is in.

Everything here is a snapshot taken at construction — the header in one round trip, the nodes in a second, and only if something asks for them. A listing walks header fields alone (state, pending, created_at), so it never pays for a thousand node records it will not render.

The stored job payload is deliberately not among the fields read. It is the largest thing on a node record, it is the one field that carries the caller's own arguments, and neither surface renders it — the node's jid is what addresses the job, and every job-inspection view already exists.

Defined Under Namespace

Classes: Node

Constant Summary collapse

NODE_FIELDS =

Every node field except payload. Ordered as Creation writes them, so the two lists read as the same record.

%w[index name class queue jid bid state deps dependents remaining pipe error].freeze
TERMINAL_STATES =

A flow is running until it is one of the other three. failed is not terminal — retrying a dead node out of the morgue puts the flow back to running — which is why it is not listed as such anywhere.

%w[succeeded abandoned].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(fid, header: nil) ⇒ Status

header: is the flow:<fid> hash already read, by a listing that fetched a page of them in one pipeline (FlowSet#snapshot); without it the header is read here.

Raises:

  • (ArgumentError) 


37
38
39
40
41
42
 
# File 'lib/wurk/flow/status.rb', line 37

def initialize(fid, header: nil)
  raise ArgumentError, 'fid required' if fid.nil? || fid.to_s.empty?

  @fid = fid.to_s
  header ? self.header = header : reload!
end

Instance Attribute Details

#fid ⇒ Object (readonly)

Returns the value of attribute fid.



32
33
34
 
# File 'lib/wurk/flow/status.rb', line 32

def fid
  @fid
end

Instance Method Details

#abandoned? ⇒ Boolean

Returns:

  • (Boolean) 


63
 
# File 'lib/wurk/flow/status.rb', line 63

def abandoned? = state == 'abandoned'

#abandoned_at ⇒ Object 



58
 
# File 'lib/wurk/flow/status.rb', line 58

def abandoned_at = numeric_or_nil(@header['abandoned_at'])

#created_at ⇒ Object 



55
 
# File 'lib/wurk/flow/status.rb', line 55

def created_at  = numeric_or_nil(@header['created_at'])

#data ⇒ Object

JSON-serializable snapshot. Node rows ride inside it because a flow without its nodes is a progress bar — the graph is the thing being asked for, and a second round of requests to assemble it would race the first.



93
94
95
96
97
98
99
100
101
 
# File 'lib/wurk/flow/status.rb', line 93

def data
  {
    'fid' => @fid, 'state' => state, 'total' => total, 'pending' => pending,
    'succeeded' => succeeded_count, 'depth' => depth, 'width' => width,
    'created_at' => created_at, 'finished_at' => finished_at, 'failed_at' => failed_at,
    'abandoned_at' => abandoned_at, 'dead_nodes' => dead_indexes,
    'nodes' => nodes.map(&:data)
  }
end

#dead_indexes ⇒ Array<Integer>

Returns the nodes the flow is failed because of — whose job reached the morgue, or whose pipe had nothing to carry. Empty for a healthy flow, and empty after abandonment, which drops the set along with the node records it points into.

Returns:

  • (Array<Integer>) —

    the nodes the flow is failed because of — whose job reached the morgue, or whose pipe had nothing to carry. Empty for a healthy flow, and empty after abandonment, which drops the set along with the node records it points into.



78
79
80
 
# File 'lib/wurk/flow/status.rb', line 78

def dead_indexes
  @dead_indexes ||= Wurk.redis { |conn| conn.call('SMEMBERS', Keys.flow_dead(@fid)) }.map(&:to_i).sort
end

#depth ⇒ Object 



52
 
# File 'lib/wurk/flow/status.rb', line 52

def depth       = @header['depth'].to_i

#exists? ⇒ Boolean

False when no flow:<fid> hash exists: a well-formed fid that was never created, or whose retention ran out. Abandoning a flow leaves the record behind on purpose, so an abandoned flow still exists here.

Returns:

  • (Boolean) 


47
 
# File 'lib/wurk/flow/status.rb', line 47

def exists? = !@header.empty?

#expiry ⇒ Object 



54
 
# File 'lib/wurk/flow/status.rb', line 54

def expiry      = @header['expiry'].to_i

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


62
 
# File 'lib/wurk/flow/status.rb', line 62

def failed?    = state == 'failed'

#failed_at ⇒ Object 



57
 
# File 'lib/wurk/flow/status.rb', line 57

def failed_at   = numeric_or_nil(@header['failed_at'])

#finished_at ⇒ Object 



56
 
# File 'lib/wurk/flow/status.rb', line 56

def finished_at = numeric_or_nil(@header['finished_at'])

#nodes ⇒ Array<Node>

Returns every node, in declaration order. Empty for an abandoned flow: the kill switch releases the node records, and their absence is the honest reading of "there is nothing left to show".

Returns:

  • (Array<Node>) —

    every node, in declaration order. Empty for an abandoned flow: the kill switch releases the node records, and their absence is the honest reading of "there is nothing left to show".



85
86
87
 
# File 'lib/wurk/flow/status.rb', line 85

def nodes
  @nodes ||= read_nodes
end

#pending ⇒ Object 



51
 
# File 'lib/wurk/flow/status.rb', line 51

def pending     = @header['pending'].to_i

#reload! ⇒ Object 



103
104
105
106
107
108
 
# File 'lib/wurk/flow/status.rb', line 103

def reload!
  self.header = Wurk.redis { |conn| conn.call('HGETALL', Keys.flow(@fid)) }
  @nodes = nil
  @dead_indexes = nil
  self
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


60
 
# File 'lib/wurk/flow/status.rb', line 60

def running?   = state == 'running'

#state ⇒ Object 



49
 
# File 'lib/wurk/flow/status.rb', line 49

def state       = @header['state']

#succeeded? ⇒ Boolean

Returns:

  • (Boolean) 


61
 
# File 'lib/wurk/flow/status.rb', line 61

def succeeded? = state == 'succeeded'

#succeeded_count ⇒ Integer

Returns nodes that have succeeded. Derived rather than stored: pending is the counter the completion script decrements, and a second field for its complement is a second thing to keep true.

Returns:

  • (Integer) —

    nodes that have succeeded. Derived rather than stored: pending is the counter the completion script decrements, and a second field for its complement is a second thing to keep true.



72
 
# File 'lib/wurk/flow/status.rb', line 72

def succeeded_count = total - pending

#terminal? ⇒ Boolean

Returns true when nothing will move this flow again. A failed flow can still recover, so it is not one of these.

Returns:

  • (Boolean) —

    true when nothing will move this flow again. A failed flow can still recover, so it is not one of these.



67
 
# File 'lib/wurk/flow/status.rb', line 67

def terminal? = TERMINAL_STATES.include?(state)

#total ⇒ Object 



50
 
# File 'lib/wurk/flow/status.rb', line 50

def total       = @header['total'].to_i

#width ⇒ Object 



53
 
# File 'lib/wurk/flow/status.rb', line 53

def width       = @header['width'].to_i