Class: SolidQueue::BlockedExecution

Inherits:
Execution show all
Defined in:
lib/solid_queue_mongoid/models/classes.rb,
lib/solid_queue_mongoid/models/blocked_execution.rb

Constant Summary collapse

INDEX_HINTS = 
{
  index_solid_queue_blocked_executions_for_release: { concurrency_key: 1, priority: 1, job_id: 1 },
  index_solid_queue_blocked_executions_for_maintenance: { expires_at: 1, concurrency_key: 1 }
}.freeze

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Execution

create_all_from_jobs, #discard, discard_all_from_jobs, discard_all_in_batches, type, #type

Methods inherited from Record

create_or_find_by!, find_by!, index, index_specifications, index_specifications=, inherited, non_blocking_lock, supports_insert_conflict_target?, transaction, use_index

Class Method Details

.create!(**attrs, &block) ⇒ Object

Make create! idempotent: if a BlockedExecution for this job already exists (e.g. created by auto-dispatch), return the existing record instead of raising.



25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/solid_queue_mongoid/models/blocked_execution.rb', line 25

def create!(**attrs, &block)
  super
rescue Mongo::Error::OperationFailure => e
  raise unless e.message.to_s.include?("E11000") || e.message.to_s.include?("duplicate key")

  job_id = attrs[:job_id] || attrs[:job]&.id
  where(job_id: job_id).first || raise(e)
rescue Mongoid::Errors::Validations => e
  return where(job_id: attrs[:job_id] || attrs[:job]&.id).first || raise(e) if uniqueness_only_error?(e.document)

  raise
end

.release_many(concurrency_keys) ⇒ Object 



59
60
61
 
# File 'lib/solid_queue_mongoid/models/blocked_execution.rb', line 59

def release_many(concurrency_keys)
  Array(concurrency_keys).count { |key| release_one(key) }
end

.release_one(concurrency_key) ⇒ Object 



63
64
65
66
67
68
69
 
# File 'lib/solid_queue_mongoid/models/blocked_execution.rb', line 63

def release_one(concurrency_key)
  # NOTE: no outer Mongoid.transaction here — #release already wraps its work
  # in a transaction. MongoDB does not support nested sessions, so wrapping
  # again would cause InvalidSessionNesting errors.
  execution = ordered.where(concurrency_key: concurrency_key).limit(1).first
  execution ? execution.release : false
end

.unblock(limit) ⇒ Object 



38
39
40
41
42
43
 
# File 'lib/solid_queue_mongoid/models/blocked_execution.rb', line 38

def unblock(limit)
  SolidQueue.instrument(:release_many_blocked, limit: limit) do |payload|
    expired_keys = expired.order(:concurrency_key).distinct(:concurrency_key).first(limit)
    payload[:size] = release_many(releasable(expired_keys))
  end
end

.unblock_all(concurrency_key, limit = nil) ⇒ Object

Convenience method: release blocked executions for a given concurrency key. limit is the maximum number to unblock (default: all).



47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/solid_queue_mongoid/models/blocked_execution.rb', line 47

def unblock_all(concurrency_key, limit = nil)
  scope = ordered.where(concurrency_key: concurrency_key)
  scope = scope.limit(limit) if limit
  count = 0
  scope.each do |execution|
    break if limit && count >= limit

    count += 1 if execution.release
  end
  count
end

Instance Method Details

#release ⇒ Object 



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/solid_queue_mongoid/models/blocked_execution.rb', line 90

def release
  SolidQueue.instrument(:release_blocked, job_id: job.id, concurrency_key: concurrency_key,
                                          released: false) do |payload|
    Mongoid.transaction do
      if acquire_concurrency_lock
        promote_to_ready
        destroy!
        payload[:released] = true
        true
      else
        false
      end
    end
  end
end

#unblock ⇒ Object 



86
87
88
 
# File 'lib/solid_queue_mongoid/models/blocked_execution.rb', line 86

def unblock
  release
end