Class: SidekiqBatch::BatchEnrollmentContext

Inherits:
Object
  • Object
show all
Defined in:
app/models/sidekiq_batch/batch_enrollment_context.rb

Overview

Scopes a block of perform_async calls to one SidekiqBatch, through a thread-local the client middleware reads on every push. Only the enrolling thread is affected; jobs a tracked worker enqueues later are not enrolled.

Defined Under Namespace

Classes: AdoptedError, AlreadyStartedError, EmptyEnrollmentError, Error, NestedError, TransactionError

Constant Summary collapse

THREAD_KEY = 
:sidekiq_batch_enrollment_context
TXN_BASELINE = 
:sidekiq_batch_enrollment_txn_baseline

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(batch) ⇒ BatchEnrollmentContext

Returns a new instance of BatchEnrollmentContext.



31
32
33
34
 
# File 'app/models/sidekiq_batch/batch_enrollment_context.rb', line 31

def initialize(batch)
  @batch          = batch
  @inserted_count = 0
end

Instance Attribute Details

#batch ⇒ Object (readonly)

Returns the value of attribute batch.



36
37
38
 
# File 'app/models/sidekiq_batch/batch_enrollment_context.rb', line 36

def batch
  @batch
end

Class Method Details

.current ⇒ Object 



23
24
25
 
# File 'app/models/sidekiq_batch/batch_enrollment_context.rb', line 23

def self.current
  Thread.current[THREAD_KEY]
end

.transaction_baseline ⇒ Object 



27
28
29
 
# File 'app/models/sidekiq_batch/batch_enrollment_context.rb', line 27

def self.transaction_baseline
  Thread.current[TXN_BASELINE] || 0
end

Instance Method Details

#enroll(payload) ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'app/models/sidekiq_batch/batch_enrollment_context.rb', line 56

def enroll(payload)
  assert_no_open_transaction!

  # The object, not its id: `belongs_to` is required by default, so an id
  # alone would make ActiveRecord SELECT the batch back on every enrolled job
  # to prove it exists. The object satisfies that check in memory.
  ::SidekiqBatchJob.create!(
    sidekiq_batch: @batch,
    jid:           payload.fetch("jid"),
    worker_class:  payload.fetch("class"),
    args:          payload.fetch("args", []),
    status:        "pending"
  )

  # Lets the server middleware recognise a tracked job without asking
  # Postgres. Sidekiq pushes only after the client chain returns, so this
  # reaches the worker. After the insert, so a failed insert stamps nothing.
  payload[PAYLOAD_BATCH_ID_KEY] = @batch.id

  @inserted_count += 1
end

#run(&block) ⇒ Object

Raises:



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'app/models/sidekiq_batch/batch_enrollment_context.rb', line 38

def run(&block)
  assert_startable!

  enroll_from(&block)

  if @inserted_count.zero?
    # Nothing to complete, so drop the row rather than leave an orphan
    # `pending` batch behind for callers who rescue the error.
    discard
    raise EmptyEnrollmentError, "jobs {} block enrolled zero jobs"
  end

  raise AdoptedError, "SidekiqBatch ##{@batch.id} was reaped as abandoned while its jobs {} block ran" \
    unless start!(enrollment_error: nil)

  complete_if_finished
end