Class: Wurk::Flow::Creation

Inherits:
Object
  • Object
show all
Includes:
JobUtil
Defined in:
lib/wurk/flow/creation.rb

Overview

Turns a validated graph into Redis state: every node's payload, every node's batch, and the queue entries for the nodes that can run now.

Two phases, and the split is the point. Everything that can be refused — a non-JSON argument, a queue-less class, a client middleware that halts the push — happens while the graph is still only in memory. Only once every node has a payload does anything reach Redis, and then all of it does, in one script (lua/flow_create.lua). A flow that half exists cannot be detected by the thing waiting on it: its parent simply never fires, silently, forever.

The accumulator is Batch::Buffer, the same one an autoflush Batch#jobs block fills, with a nil threshold — a batch buffers to bound its pipeline and may flush early, whereas a flow has no legal partial flush, and nil is how that is spelled.

Payloads are built the way Client#push builds one and for the same reasons: JobUtil normalizes and verifies, then the client middleware chain runs. Every node's job goes through it — including the ones this write only records — because a chain that ran for some nodes and not others would produce a graph whose payloads disagree about what a job of this app looks like.

Constant Summary collapse

WAITING =

state on a node record. A node is waiting until every dependency has succeeded; enqueued once its payload is on a queue. Creation writes both — roots are enqueued, the rest wait — and the terminal states belong to the completion step that owns those transitions.

'waiting'
ENQUEUED = 
'enqueued'
PIPE_SENTINEL_PREFIX =

What a piped node carries in place of the argument it does not have yet. The upstream result is not known until the upstream runs, so the payload is stored with this sentinel where the argument goes and lua/flow_advance.lua swaps the two bytes-for-bytes on the way to the queue — no cjson round trip, which would round a 64-bit number in some other argument to a double.

It carries the fid and the node index so a sentinel is unique to the one splice it belongs to, and so a payload caught in the dashboard mid flow says what it is waiting for rather than showing a bare marker.

'wurk.flow.pipe:'

Constants included from JobUtil

JobUtil::BOUND_OPTIONS, JobUtil::RETRY_FOR_MAX, JobUtil::TRACK_VALUES, JobUtil::TRANSIENT_ATTRIBUTES

Instance Method Summary collapse

Methods included from JobUtil

#normalize_item, #now_in_millis, positive_seconds?, scheduled_member, #validate, validate_bounds!, validate_track!, #verify_json

Constructor Details

#initialize(flow, config: nil) ⇒ Creation

Returns a new instance of Creation.



54
55
56
57
 
# File 'lib/wurk/flow/creation.rb', line 54

def initialize(flow, config: nil)
  @flow   = flow
  @config = config || Wurk.configuration
end

Instance Method Details

#call ⇒ Array<Hash>

Returns every node's payload, in node-index order. The graph is in Redis by the time this returns.

Returns:

  • (Array<Hash>) —

    every node's payload, in node-index order. The graph is in Redis by the time this returns.

Raises:



62
63
64
65
66
67
 
# File 'lib/wurk/flow/creation.rb', line 62

def call
  payloads = build_payloads
  write(payloads)
  emit_enqueued(payloads)
  payloads
end