Class: Wurk::Flow

Inherits:
Object
  • Object
show all
Defined in:
lib/wurk/flow.rb,
lib/wurk/flow/node.rb,
lib/wurk/flow/chain.rb,
lib/wurk/flow/status.rb,
lib/wurk/flow/builder.rb,
lib/wurk/flow/creation.rb,
lib/wurk/flow/completion.rb

Overview

A DAG of jobs: fan out, fan in, and run each node only after everything it depends on has succeeded. A Wurk extra — Sidekiq Pro has batches, which are a flow exactly one level deep.

A node is one job, wrapped in its own batch. That is the whole model: a batch already owns success callbacks, the death cascade, subtree gating and a rendered status, so a flow is the parent relation between batches and not a second completion tracker. Two trackers that can disagree is the failure mode of this feature.

Work that is wide rather than deep belongs inside a node — that node's job opens its own batch and enqueues however many children it likes, and batch nesting already blocks the node on the whole subtree. Sibling nodes are for work with different shapes, not for parallelism.

Constructing a flow validates it and writes nothing. Every refusal happens here, before Redis has heard of the graph: a cycle that would deadlock silently, a name nothing declares, a graph past the caps below.

Required eagerly from wurk.rb, unlike API and Telemetry which load on opt-in: those buy back real pre-fork heap (1.2 MB) or depend on a gem that may be absent, while this is three files of pure Ruby whose load sits inside the run-to-run spread of require "wurk" (~250 ms ± 30). Lazy-loading it would also put the Sidekiq::Flow alias behind a first use.

Examples:

A diamond — C runs once, after both A and B succeed

Wurk::Flow.new do |f|
  a = f.job(FetchJob, 'https://a')
  b = f.job(FetchJob, 'https://b')
  f.job(MergeJob, depends_on: [a, b])
end.run

The same graph, addressed by name — forward references are legal

Wurk::Flow.new do |f|
  f.job(MergeJob, name: :merge, depends_on: %i[a b])
  f.job(FetchJob, 'https://a', name: :a)
  f.job(FetchJob, 'https://b', name: :b)
end

A chain — every step is handed the step before it

Wurk::Flow.chain do |c|
  c.job(FetchJob, 'https://example.com')
  c.job(ParseJob)
  c.job(StoreJob, 'reports')
end.run

See Also:

Defined Under Namespace

Classes: Builder, Chain, Completion, Creation, CycleError, InvalidGraph, LimitExceeded, Node, Status

Constant Summary collapse

FID_BYTES =

Same generator and length as a batch's bid: URL-safe base64 of 10 random bytes. A flow id is handled by the same code paths (dashboard routes, API captures, log lines) and there is no reason for it to look different.

10
MAX_NODES =

Total nodes. Creation is one atomic write, and Redis is single threaded — the whole graph is that write's payload, and each node costs a batch's keys. Beyond this the graph is not big, it is the wrong shape.

1_000
MAX_DEPTH =

Longest dependency path. Each level costs a full callback hop — a job finishing, a callback job enqueued, fetched and run — so depth is latency; it is also how deep the batch death cascade recurses.

50
MAX_WIDTH =

The most edges on either side of one node. Fan-in bounds how many siblings race to advance the same parent; fan-out bounds how many nodes a single completion enqueues.

100
COMPLETION_CALLBACK =

The callback class every node's batch is created with, for both events a node can reach: :success advances the flow, :death marks it failed. Named here because creation writes the spec and Completion is what answers to it; a literal in both places is a flow that stalls silently the day one of them is renamed.

'Wurk::Flow::Completion'

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(&block) ⇒ Flow

Returns a new instance of Flow.

Raises:

  • (ArgumentError) 


192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/wurk/flow.rb', line 192

def initialize(&block)
  raise ArgumentError, 'flow requires a block' unless block

  @fid = SecureRandom.urlsafe_base64(FID_BYTES)
  builder = Builder.new
  block.call(builder)
  @nodes = builder.build.freeze
  @depth = builder.depth
  @width = builder.width
  @expiry = Wurk::Batch::DEFAULT_EXPIRY_SECONDS
  @callback_queue = 'default'
  @created = false
end

Instance Attribute Details

#bids ⇒ Array<String>? (readonly)

Returns each node's bid, in node-index order, or nil before #run. A node's batch is what carries its completion.

Returns:

  • (Array<String>, nil) —

    each node's bid, in node-index order, or nil before #run. A node's batch is what carries its completion.



123
124
125
 
# File 'lib/wurk/flow.rb', line 123

def bids
  @bids
end

#callback_queue ⇒ Object

The queue every node batch's callbacks run on. Same knob and same default as Batch#callback_queue: a flow advances one callback at a time, so a host that keeps callbacks off a saturated default needs to be able to say so here too.



129
130
131
 
# File 'lib/wurk/flow.rb', line 129

def callback_queue
  @callback_queue
end

#depth ⇒ Integer (readonly)

Returns number of levels — 1 for a flow with no edges.

Returns:

  • (Integer) —

    number of levels — 1 for a flow with no edges.



107
108
109
 
# File 'lib/wurk/flow.rb', line 107

def depth
  @depth
end

#expiry ⇒ Integer (readonly)

Returns seconds every key this flow creates lives for, absent the shorter clocks a finished batch stamps on its own keys.

Returns:

  • (Integer) —

    seconds every key this flow creates lives for, absent the shorter clocks a finished batch stamps on its own keys.



114
115
116
 
# File 'lib/wurk/flow.rb', line 114

def expiry
  @expiry
end

#fid ⇒ String (readonly)

Returns this flow's id. Allocated here and written to Redis by creation, the same way Batch allocates a bid without a round trip — an unrun flow costs nothing but the object.

Returns:

  • (String) —

    this flow's id. Allocated here and written to Redis by creation, the same way Batch allocates a bid without a round trip — an unrun flow costs nothing but the object.



101
102
103
 
# File 'lib/wurk/flow.rb', line 101

def fid
  @fid
end

#jids ⇒ Array<String>? (readonly)

Returns each node's jid, in node-index order, or nil before #run. A node is one job (decision 0), so this is the address of its result: Wurk::Status.get(flow.jids[i]).

Returns:

  • (Array<String>, nil) —

    each node's jid, in node-index order, or nil before #run. A node is one job (decision 0), so this is the address of its result: Wurk::Status.get(flow.jids[i]).



119
120
121
 
# File 'lib/wurk/flow.rb', line 119

def jids
  @jids
end

#nodes ⇒ Array<Flow::Node> (readonly)

Returns every node, frozen, in topological order.

Returns:

  • (Array<Flow::Node>) —

    every node, frozen, in topological order.



104
105
106
 
# File 'lib/wurk/flow.rb', line 104

def nodes
  @nodes
end

#width ⇒ Integer (readonly)

Returns the most edges on either side of any one node.

Returns:

  • (Integer) —

    the most edges on either side of any one node.



110
111
112
 
# File 'lib/wurk/flow.rb', line 110

def width
  @width
end

Class Method Details

.abandon(fid) ⇒ Boolean

The kill switch: give up on a flow and release what it is holding.

Marks the flow abandoned and drops every node record, every node batch and the dead-node set — the ~9 keys per node that are the actual weight. The flow's own record stays, on the clock it was created with, because a flow that vanishes silently is indistinguishable from one that was never created; abandoned is the answer to "where did it go".

It does not chase in-flight jobs out of their queues. Same caveat and same wording as Batch::Status#delete: jobs already queued will run and ack against a batch that is gone. Nothing they do can revive the flow — every write in Completion claims on a node record this released.

Only a live flow can be abandoned. A flow that already succeeded, or that was already abandoned, is not stuck and is left exactly as it is.

idempotent: true is that claim spent: a replay after a lost reply finds the flow already terminal and writes nothing, so the pool may retry a connection error rather than give up on the kill switch. The cost is that such a replay reports false for a call that did apply — the flow record is the authority, not the return value.

Returns:

  • (Boolean) —

    true when this call abandoned the flow.



179
180
181
182
183
184
185
186
187
188
189
 
# File 'lib/wurk/flow.rb', line 179

def abandon(fid) # rubocop:disable Naming/PredicateMethod
  now = ::Process.clock_gettime(::Process::CLOCK_REALTIME).to_s
  released = Wurk.redis(idempotent: true) do |conn|
    Wurk::Lua::Loader.eval_cached(
      conn, :flow_abandon,
      keys: [Keys.flow(fid), 'batches', 'dead-batches'],
      argv: [now, *Wurk::Batch::KEY_SUFFIXES]
    )
  end
  released.to_i >= 0
end

.chain(&block) ⇒ Object

A flow with one path through it: every step waits on the step before it and is handed that step's stored return value as its last argument.

Wurk::Flow.chain do |c|
c.job(FetchJob, 'https://example.com')
c.job(ParseJob)                        # ParseJob#perform(fetch_result)
c.job(StoreJob, 'reports')             # StoreJob#perform('reports', parse_result)
end.run

Every step but the last is enqueued with track: true, because the pipe carries Status's stored result and tracking is opt-in per job. A result too big to store — past Middleware::Status::MAX_RESULT_BYTES — stops the chain and fails the flow rather than piping the truncated head: a shortened display is lossy, a shortened argument is wrong.

Raises:

  • (ArgumentError)

See Also:



149
150
151
152
153
 
# File 'lib/wurk/flow.rb', line 149

def chain(&block)
  raise ArgumentError, 'flow requires a block' unless block

  new { |builder| block.call(Chain.new(builder)) }
end

Instance Method Details

#abandon ⇒ Boolean

Returns true when this call abandoned the flow.

Returns:

  • (Boolean) —

    true when this call abandoned the flow.

See Also:



246
 
# File 'lib/wurk/flow.rb', line 246

def abandon = self.class.abandon(@fid)

#created? ⇒ Boolean

Returns:

  • (Boolean) 


221
 
# File 'lib/wurk/flow.rb', line 221

def created? = @created

#expires_in(duration) ⇒ Object

Retention for every key this flow creates, its nodes' batches included. The batch clock (30 days) by default, because a flow is a relation between batches and an abandoned one has to disappear the same way they do — on its own, with no sweeper to elect a leader for.



216
217
218
219
 
# File 'lib/wurk/flow.rb', line 216

def expires_in(duration)
  @expiry = duration.to_i
  self
end

#roots ⇒ Array<Flow::Node>

Returns the nodes with no dependencies — what creation enqueues immediately; everything else waits on a callback.

Returns:

  • (Array<Flow::Node>) —

    the nodes with no dependencies — what creation enqueues immediately; everything else waits on a callback.



210
 
# File 'lib/wurk/flow.rb', line 210

def roots = @nodes.select(&:root?)

#run ⇒ self

Create the flow: persist the whole graph and enqueue the nodes that have nothing to wait for, in one atomic write. Nothing before this touches Redis, and a refusal here leaves nothing behind.

The claim is taken before the write, not after it. A write whose reply is lost may well have applied, and a second #run would build fresh jids only to be turned away by the script's own claim — leaving this object describing a flow that Redis does not have. One flow object creates one flow; recovering from an ambiguous failure means building another.

Returns:

  • (self) 


234
235
236
237
238
239
240
241
242
 
# File 'lib/wurk/flow.rb', line 234

def run
  raise "flow #{@fid} has already been created" if @created

  @created = true
  payloads = Creation.new(self).call
  @jids = payloads.map { |payload| payload['jid'] }.freeze
  @bids = payloads.map { |payload| payload['bid'] }.freeze
  self
end

#size ⇒ Object 



206
 
# File 'lib/wurk/flow.rb', line 206

def size = @nodes.size