Class: JobWorkflow::Context

Inherits:
Object
  • Object
show all
Defined in:
lib/job_workflow/context.rb,
sig/generated/job_workflow/context.rbs

Overview

rubocop:disable Metrics/ClassLength

Constant Summary collapse

EACH_TASK_CURSOR_MARKER =

rubocop:disable Metrics/ClassLength

Returns:

  • (::String) 
"__job_workflow_each_cursor__"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(workflow:, arguments:, task_context:, output:, job_status:, job: nil) ⇒ Context

: ( workflow: Workflow, arguments: Arguments, task_context: TaskContext, output: Output, job_status: JobStatus, ?job: _JobInterface? ) -> void

Parameters:



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/job_workflow/context.rb', line 57

def initialize(workflow:, arguments:, task_context:, output:, job_status:, job: nil) # rubocop:disable Metrics/ParameterLists, Metrics/AbcSize, Metrics/MethodLength
  if job&.class.respond_to?(:_workflow) && job.class._workflow != workflow
    raise "job does not match the provided workflow"
  end

  self.job = job
  self.workflow = workflow
  self.arguments = arguments
  self.task_context = task_context
  self.output = output
  self.job_status = job_status
  self.enabled_with_each_value = false
  self.throttle_index = 0
  self.skip_in_dry_run_index = 0
  self.current_step = nil
  self.current_cursor = nil
end

Instance Attribute Details

#arguments ⇒ Arguments

Signature:

  • Arguments

Returns:



8
9
10
 
# File 'lib/job_workflow/context.rb', line 8

def arguments
  @arguments
end

#current_cursor ⇒ Object

Signature:

  • untyped

Returns:

  • (Object) 


288
289
290
 
# File 'lib/job_workflow/context.rb', line 288

def current_cursor
  @current_cursor
end

#current_step ⇒ ActiveJob::Continuation::Step?

Signature:

  • ActiveJob::Continuation::Step?

Returns:

  • (ActiveJob::Continuation::Step, nil) 


287
288
289
 
# File 'lib/job_workflow/context.rb', line 287

def current_step
  @current_step
end

#enabled_with_each_value ⇒ Boolean

Signature:

  • bool

Returns:

  • (Boolean) 


284
285
286
 
# File 'lib/job_workflow/context.rb', line 284

def enabled_with_each_value
  @enabled_with_each_value
end

#job ⇒ _JobInterface?

Signature:

  • _JobInterface?

Returns:

  • (_JobInterface, nil) 


278
279
280
 
# File 'lib/job_workflow/context.rb', line 278

def job
  @job
end

#job_status ⇒ JobStatus

Signature:

  • JobStatus

Returns:



10
11
12
 
# File 'lib/job_workflow/context.rb', line 10

def job_status
  @job_status
end

#output ⇒ Output

Signature:

  • Output

Returns:



9
10
11
 
# File 'lib/job_workflow/context.rb', line 9

def output
  @output
end

#skip_in_dry_run_index ⇒ Integer

Signature:

  • Integer

Returns:

  • (Integer) 


286
287
288
 
# File 'lib/job_workflow/context.rb', line 286

def skip_in_dry_run_index
  @skip_in_dry_run_index
end

#task_context ⇒ TaskContext

Signature:

  • TaskContext

Returns:



283
284
285
 
# File 'lib/job_workflow/context.rb', line 283

def task_context
  @task_context
end

#throttle_index ⇒ Integer

Signature:

  • Integer

Returns:

  • (Integer) 


285
286
287
 
# File 'lib/job_workflow/context.rb', line 285

def throttle_index
  @throttle_index
end

#workflow ⇒ Workflow

Signature:

  • Workflow

Returns:



7
8
9
 
# File 'lib/job_workflow/context.rb', line 7

def workflow
  @workflow
end

Class Method Details

.deserialize(hash) ⇒ Context

: (Hash[String, untyped]) -> Context

Parameters:

  • (Hash[String, untyped])

Returns:



27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/job_workflow/context.rb', line 27

def deserialize(hash)
  workflow = hash.fetch("workflow")
  new(
    job: hash["job"],
    workflow: hash.fetch("workflow"),
    arguments: Arguments.new(data: workflow.build_arguments_hash),
    task_context: TaskContext.deserialize(
      hash["task_context"].merge(
        "task" => workflow.fetch_task(
          hash.fetch(
            "task_context",
            {} #: Hash[String, untyped]
          )["task_name"]&.to_sym
        )
      )
    ),
    output: Output.deserialize(hash),
    job_status: JobStatus.deserialize(hash)
  )
end

.from_hash(hash) ⇒ Context

: (Hash[Symbol, untyped]) -> Context

Parameters:

  • (Hash[Symbol, untyped])

Returns:



14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/job_workflow/context.rb', line 14

def from_hash(hash)
  workflow = hash.fetch(:workflow)
  new(
    job: hash[:job],
    workflow:,
    arguments: Arguments.new(data: workflow.build_arguments_hash),
    task_context: TaskContext.new(**(hash[:task_context] || {}).symbolize_keys),
    output: Output.from_hash_array(hash.fetch(:task_outputs, [])),
    job_status: JobStatus.from_hash_array(hash.fetch(:task_job_statuses, []))
  )
end

Instance Method Details

#_add_task_output(task_output) ⇒ void

This method returns an undefined value.

: (TaskOutput) -> void

Parameters:



263
264
265
 
# File 'lib/job_workflow/context.rb', line 263

def _add_task_output(task_output)
  output.add_task_output(task_output)
end

#_job ⇒ _JobInterface?

: () -> _JobInterface?

Returns:

  • (_JobInterface, nil) 


117
118
119
 
# File 'lib/job_workflow/context.rb', line 117

def _job
  job
end

#_job=(job) ⇒ void

This method returns an undefined value.

: (_JobInterface) -> void

Parameters:

  • (_JobInterface) 


112
113
114
 
# File 'lib/job_workflow/context.rb', line 112

def _job=(job)
  self.job = job
end

#_load_parent_task_output ⇒ void

This method returns an undefined value.

: () -> void



268
269
270
271
272
273
274
 
# File 'lib/job_workflow/context.rb', line 268

def _load_parent_task_output
  return unless sub_job?

  workflow_status = WorkflowStatus.find(parent_job_id)
  parent_context = workflow_status.context
  parent_context.output.flat_task_outputs.each { |task_output| output.add_task_output(task_output) }
end

#_task_context ⇒ TaskContext

: () -> TaskContext

Returns:



246
247
248
 
# File 'lib/job_workflow/context.rb', line 246

def _task_context
  task_context
end

#_update_arguments(other_arguments) ⇒ Context

: (Hash[Symbol, untyped]) -> Context

Parameters:

  • (Hash[Symbol, untyped])

Returns:



81
82
83
84
 
# File 'lib/job_workflow/context.rb', line 81

def _update_arguments(other_arguments)
  self.arguments = arguments.merge(other_arguments.symbolize_keys)
  self
end

#_with_current_step(step, cursor: nil) { ... } ⇒ void

This method returns an undefined value.

: (ActiveJob::Continuation::Step, ?cursor: untyped) { () -> void } -> void

Parameters:

  • (ActiveJob::Continuation::Step)
  • cursor: (Object) (defaults to: nil)

Yields:

Yield Returns:

  • (void) 


251
252
253
254
255
256
257
258
259
260
 
# File 'lib/job_workflow/context.rb', line 251

def _with_current_step(step, cursor: nil)
  previous_step = current_step
  previous_cursor = current_cursor
  self.current_step = step
  self.current_cursor = cursor
  yield
ensure
  self.current_step = previous_step
  self.current_cursor = previous_cursor
end

#_with_each_value(task, start_index: nil) ⇒ Enumerator[Context]

: (Task, ?start_index: Integer?) -> Enumerator

Parameters:

  • (Task)
  • start_index: (Integer, nil) (defaults to: nil)

Returns:



142
143
144
145
146
147
148
149
150
151
 
# File 'lib/job_workflow/context.rb', line 142

def _with_each_value(task, start_index: nil)
  raise "Nested _with_each_value calls are not allowed" if enabled_with_each_value

  self.enabled_with_each_value = true
  Enumerator.new do |y|
    with_task_context(task, y, start_index:)
  ensure
    self.enabled_with_each_value = false
  end
end

#_with_task_throttle { ... } ⇒ void

This method returns an undefined value.

: () { () -> void } -> void

Yields:

Yield Returns:

  • (void) 


154
155
156
157
158
159
160
161
 
# File 'lib/job_workflow/context.rb', line 154

def _with_task_throttle(&)
  task = task_context.task || (raise "with_throttle can be called only within iterate_each_value")

  semaphore = task.throttle.semaphore
  return yield if semaphore.nil?

  semaphore.with(&)
end

#build_step_cursor(value) ⇒ Object

: (untyped) -> untyped

Parameters:

  • (Object)

Returns:

  • (Object) 


302
303
304
305
306
307
308
309
310
311
 
# File 'lib/job_workflow/context.rb', line 302

def build_step_cursor(value)
  return value unless each_task?
  return task_context.index if value.nil?

  {
    EACH_TASK_CURSOR_MARKER => true,
    "index" => task_context.index,
    "cursor" => value
  }
end

#calculate_dry_run(task) ⇒ Boolean

: (Task) -> bool

Parameters:

Returns:

  • (Boolean) 


409
410
411
 
# File 'lib/job_workflow/context.rb', line 409

def calculate_dry_run(task)
  workflow.dry_run_config.evaluate(self) || task.dry_run_config.evaluate(self)
end

#checkpoint! ⇒ void

This method returns an undefined value.

: () -> void



103
104
105
106
107
108
109
 
# File 'lib/job_workflow/context.rb', line 103

def checkpoint!
  step = current_step || (raise "checkpoint! can be called only in task")

  return step.checkpoint! unless each_task?

  step.set!(build_step_cursor(current_cursor))
end

#clear_after_each_index_and_value ⇒ void

This method returns an undefined value.

: () -> void



370
371
372
373
 
# File 'lib/job_workflow/context.rb', line 370

def clear_after_each_index_and_value
  self.throttle_index = 0
  self.skip_in_dry_run_index = 0
end

#concurrency_key ⇒ String?

: () -> String?

Returns:

  • (String, nil) 


134
135
136
137
138
139
 
# File 'lib/job_workflow/context.rb', line 134

def concurrency_key
  task = task_context.task
  return if task.nil?

  [task_context.parent_job_id, task.task_name].compact.join("/")
end

#cursor ⇒ Object

: () -> untyped

Returns:

  • (Object) 


87
88
89
90
91
 
# File 'lib/job_workflow/context.rb', line 87

def cursor
  return if current_step.nil?

  current_cursor
end

#dry_run? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


211
212
213
 
# File 'lib/job_workflow/context.rb', line 211

def dry_run?
  task_context.dry_run
end

#each_task? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


296
297
298
299
 
# File 'lib/job_workflow/context.rb', line 296

def each_task?
  task = task_context.task || (raise "each_task? can be called only in task")
  task.each?
end

#each_task_output ⇒ TaskOutput?

: () -> TaskOutput?

Returns:



235
236
237
238
239
240
241
242
243
 
# File 'lib/job_workflow/context.rb', line 235

def each_task_output
  task = task_context.task
  raise "each_task_output can be called only _with_task block" if task.nil?
  raise "each_task_output can be called only _with_each_value block" unless task_context.enabled?

  task_name = task.task_name
  each_index = task_context.index
  output.fetch(task_name:, each_index:)
end

#each_value ⇒ Object

: () -> untyped

Returns:

  • (Object) 


228
229
230
231
232
 
# File 'lib/job_workflow/context.rb', line 228

def each_value
  raise "each_value can be called only within each_values block" unless task_context.enabled?

  task_context.value
end

#instrument(operation = "custom", **payload) { ... } ⇒ Object

Instruments a custom operation with ActiveSupport::Notifications. This creates a span in OpenTelemetry (if enabled) and logs the event.

: (?String, **untyped) { () -> untyped } -> untyped

Examples:

Basic usage

```ruby
ctx.instrument("api_call", endpoint: "/users") do
  HTTP.get("https://api.example.com/users")
end
```

With automatic operation name

```ruby
ctx.instrument do
  # operation name defaults to "custom"
  expensive_operation()
end
```

Parameters:

  • (String)
  • (Object)

Yields:

Yield Returns:

  • (Object)

Returns:

  • (Object) 


197
198
199
200
201
202
203
204
205
206
207
208
 
# File 'lib/job_workflow/context.rb', line 197

def instrument(operation = "custom", **payload, &)
  task = task_context.task
  full_payload = {
    job_id: job_id,
    job_name: job.class.name,
    task_name: task&.task_name,
    each_index: task_context.index,
    operation:,
    **payload
  }
  Instrumentation.instrument_custom(operation, full_payload, &)
end

#job_id ⇒ String

: () -> String

Returns:

  • (String) 


122
123
124
125
126
 
# File 'lib/job_workflow/context.rb', line 122

def job_id
  local_job = job || (raise "job is not set")

  local_job.job_id
end

#parent_job_id ⇒ String

: () -> String

Returns:

  • (String) 


291
292
293
 
# File 'lib/job_workflow/context.rb', line 291

def parent_job_id
  _task_context.parent_job_id || job_id
end

#reset_task_context_if_task_changed(task) ⇒ void

This method returns an undefined value.

: (Task) -> void

Parameters:



350
351
352
353
354
 
# File 'lib/job_workflow/context.rb', line 350

def reset_task_context_if_task_changed(task)
  return if sub_job?

  self.task_context = TaskContext.new if task_context.task&.task_name != task.task_name
end

#serialize ⇒ Hash[String, untyped]

: () -> Hash[String, untyped]

Returns:

  • (Hash[String, untyped]) 


76
77
78
 
# File 'lib/job_workflow/context.rb', line 76

def serialize
  sub_job? ? serialize_for_sub_job : serialize_for_job
end

#serialize_for_job ⇒ Hash[String, untyped]

: () -> Hash[String, untyped]

Returns:

  • (Hash[String, untyped]) 


314
315
316
317
318
319
320
 
# File 'lib/job_workflow/context.rb', line 314

def serialize_for_job
  {
    "task_context" => _task_context.serialize,
    "task_outputs" => output.flat_task_outputs.map(&:serialize),
    "task_job_statuses" => job_status.flat_task_job_statuses.map(&:serialize)
  }
end

#serialize_for_sub_job ⇒ Hash[String, untyped]

: () -> Hash[String, untyped]

Returns:

  • (Hash[String, untyped]) 


323
324
325
326
327
328
329
330
 
# File 'lib/job_workflow/context.rb', line 323

def serialize_for_sub_job
  task_output = output.fetch(task_name: task_context.task&.task_name, each_index: task_context.index)
  {
    "task_context" => _task_context.serialize,
    "task_outputs" => [task_output].compact.map(&:serialize),
    "task_job_statuses" => []
  }
end

#set_cursor!(value) ⇒ void

This method returns an undefined value.

: (untyped) -> void

Parameters:

  • (Object) 


94
95
96
97
98
99
100
 
# File 'lib/job_workflow/context.rb', line 94

def set_cursor!(value)
  step = current_step || (raise "set_cursor! can be called only in task")

  ActiveJob::Arguments.serialize([value])
  self.current_cursor = value
  step.set!(build_step_cursor(value))
end

#skip_in_dry_run(dry_run_name = nil, fallback: nil) { ... } ⇒ Object

: (?Symbol?, ?fallback: untyped) { () -> untyped } -> untyped

Parameters:

  • (Symbol, nil)
  • fallback: (Object) (defaults to: nil)

Yields:

Yield Returns:

  • (Object)

Returns:

  • (Object) 


216
217
218
219
220
221
222
223
224
225
 
# File 'lib/job_workflow/context.rb', line 216

def skip_in_dry_run(dry_run_name = nil, fallback: nil)
  local_job = job || (raise "job is not set")
  task_context.task || (raise "skip_in_dry_run can be called only within with_task_context")

  current_index = skip_in_dry_run_index
  self.skip_in_dry_run_index += 1
  Instrumentation.instrument_dry_run(local_job, self, dry_run_name, current_index, dry_run?) do
    dry_run? ? fallback : yield
  end
end

#sub_job? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


129
130
131
 
# File 'lib/job_workflow/context.rb', line 129

def sub_job?
  parent_job_id != job_id
end

#throttle(limit:, key: nil, ttl: 180) { ... } ⇒ void

This method returns an undefined value.

: (limit: Integer, ?key: String?, ?ttl: Integer) { () -> void } -> void

Parameters:

  • limit: (Integer)
  • key: (String, nil) (defaults to: nil)
  • ttl: (Integer) (defaults to: 180)

Yields:

Yield Returns:

  • (void) 


164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/job_workflow/context.rb', line 164

def throttle(limit:, key: nil, ttl: 180, &)
  task = task_context.task || (raise "throttle can be called only in task")

  semaphore = Semaphore.new(
    concurrency_key: key || "#{task.throttle_prefix_key}:#{throttle_index}",
    concurrency_limit: limit,
    concurrency_duration: ttl.seconds
  )

  self.throttle_index += 1

  semaphore.with(&)
end

#wait_next_retry(task, task_retry, next_retry_count, error) ⇒ void

This method returns an undefined value.

: (Task, TaskRetry, Integer, StandardError) -> void

Parameters:



402
403
404
405
406
 
# File 'lib/job_workflow/context.rb', line 402

def wait_next_retry(task, task_retry, next_retry_count, error)
  delay = task_retry.delay_for(next_retry_count)
  Instrumentation.notify_task_retry(task, self, job_id, next_retry_count, delay, error)
  sleep(delay)
end

#with_each_index_and_value(task, start_index: nil) {|arg0, arg1| ... } ⇒ void

This method returns an undefined value.

: (Task, ?start_index: Integer?) { (untyped, Integer) -> void } -> void

Parameters:

  • (Task)
  • start_index: (Integer, nil) (defaults to: nil)

Yields:

Yield Parameters:

  • arg0 (Object)
  • arg1 (Integer)

Yield Returns:

  • (void) 


357
358
359
360
361
362
363
364
365
366
367
 
# File 'lib/job_workflow/context.rb', line 357

def with_each_index_and_value(task, start_index: nil)
  resume_index = start_index || task_context.index

  task.each.call(self).each.with_index do |value, index|
    next if index < resume_index

    yield value, index

    break if sub_job?
  end
end

#with_retry(task) {|arg0| ... } ⇒ void

This method returns an undefined value.

: (Task) { (Integer) -> void } -> void

Parameters:

Yields:

Yield Parameters:

  • arg0 (Integer)

Yield Returns:

  • (void) 


386
387
388
389
390
391
392
393
394
395
396
397
398
399
 
# File 'lib/job_workflow/context.rb', line 386

def with_retry(task)
  task_retry = task.task_retry
  0.upto(task_retry.count) do |retry_count|
    next if retry_count < task_context.retry_count

    yield retry_count
    break
  rescue StandardError => e
    next_retry_count = retry_count + 1
    raise e if next_retry_count >= task_retry.count

    wait_next_retry(task, task_retry, next_retry_count, e)
  end
end

#with_task_context(task, yielder, start_index: nil) ⇒ void

This method returns an undefined value.

: (Task, Enumerator::Yielder, ?start_index: Integer?) -> void

Parameters:

  • (Task)
  • (Enumerator::Yielder)
  • start_index: (Integer, nil) (defaults to: nil) 


333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
 
# File 'lib/job_workflow/context.rb', line 333

def with_task_context(task, yielder, start_index: nil) # rubocop:disable Metrics/MethodLength
  reset_task_context_if_task_changed(task)

  with_each_index_and_value(task, start_index:) do |value, index|
    dry_run = calculate_dry_run(task)
    with_retry(task) do |retry_count|
      self.task_context = TaskContext.new(task:, parent_job_id:, index:, value:, retry_count:, dry_run:)
      with_task_timeout do
        yielder << self
      end
    end
  ensure
    clear_after_each_index_and_value
  end
end

#with_task_timeout { ... } ⇒ void

This method returns an undefined value.

: () { () -> void } -> void

Yields:

Yield Returns:

  • (void) 


376
377
378
379
380
381
382
383
 
# File 'lib/job_workflow/context.rb', line 376

def with_task_timeout
  task = task_context.task || (raise "with_task_timeout can be called only within with_task_context")

  timeout = task.timeout
  return yield if timeout.nil?

  Timeout.timeout(timeout) { yield } # rubocop:disable Style/ExplicitBlockArgument
end