Class: JobWorkflow::QueueAdapters::SolidQueueAdapter

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

Overview

rubocop:disable Naming/PredicateMethod, Metrics/ClassLength

Defined Under Namespace

Modules: ClaimedExecutionPatch, SchedulingPatch

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ SolidQueueAdapter

Note:
  • Registry scope: @semaphore_registry is process-scoped (shared across fibers/threads in the same process) and lives for the lifetime of the worker process. It is not serialized to persistent storage; semaphores are transient per worker instance.
  • Cleanup: The adapter relies on SolidQueue::Worker lifecycle hooks to clean up active semaphores when the worker stops. If a worker crashes, semaphores will leak until the underlying database records expire or are manually cleaned.

: () -> void



16
17
18
19
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 16

def initialize
  @semaphore_registry = {} #: Hash[Object, ^(SolidQueue::Worker) -> void]
  super
end

Instance Attribute Details

#semaphore_registry ⇒ Hash[Object, ^(SolidQueue::Worker) -> void] (readonly)

Signature:

  • Hash[Object, ^(SolidQueue::Worker) -> void]

Returns:

  • (Hash[Object, ^(SolidQueue::Worker) -> void]) 


221
222
223
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 221

def semaphore_registry
  @semaphore_registry
end

Instance Method Details

#build_page(jobs, limit:) ⇒ Hash[Symbol, untyped]

: (Array, limit: Integer) -> Hash[Symbol, untyped]

Parameters:

  • (Array[SolidQueue::Job])
  • limit: (Integer)

Returns:

  • (Hash[Symbol, untyped]) 


255
256
257
258
259
260
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 255

def build_page(jobs, limit:)
  {
    jobs: jobs.first(limit).map { |job| normalized_job_data(job) },
    next_cursor: jobs.size > limit ? jobs[limit - 1].id.to_s : nil
  }
end

#clear_queue(queue_name) ⇒ Boolean

: (String) -> bool

Parameters:

  • (String)

Returns:

  • (Boolean) 


145
146
147
148
149
150
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 145

def clear_queue(queue_name)
  return false unless defined?(SolidQueue::Queue)

  SolidQueue::Queue.find_by_name(queue_name).clear
  true
end

#fetch_job_contexts(job_ids) ⇒ Array[Hash[String, untyped]]

Note:
  • Fetches job_workflow_context hashes for the given job IDs.

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

Parameters:

  • (Array[String])

Returns:

  • (Array[Hash[String, untyped]]) 


180
181
182
183
184
185
186
187
188
189
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 180

def fetch_job_contexts(job_ids)
  return [] unless defined?(SolidQueue::Job)
  return [] if job_ids.empty?

  jobs = without_query_cache { SolidQueue::Job.where(active_job_id: job_ids).to_a }
  jobs.filter_map do |job|
    args = job.arguments
    args.is_a?(Hash) ? args["job_workflow_context"] : nil
  end
end

#fetch_job_statuses(job_ids) ⇒ Hash[String, untyped]

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

Parameters:

  • (Array[String])

Returns:

  • (Hash[String, untyped]) 


74
75
76
77
78
79
80
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 74

def fetch_job_statuses(job_ids)
  return {} unless defined?(SolidQueue::Job)

  without_query_cache do
    SolidQueue::Job.where(active_job_id: job_ids).index_by(&:active_job_id)
  end
end

#fetch_root_workflow_job_page(job_class_name:, limit:, cursor:) ⇒ Hash[Symbol, untyped]

: (job_class_name: String, limit: Integer, cursor: String?) -> Hash[Symbol, untyped]

Parameters:

  • job_class_name: (String)
  • limit: (Integer)
  • cursor: (String, nil)

Returns:

  • (Hash[Symbol, untyped]) 


167
168
169
170
171
172
173
174
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 167

def fetch_root_workflow_job_page(job_class_name:, limit:, cursor:)
  return { jobs: [], next_cursor: nil } unless defined?(SolidQueue::Job)

  without_query_cache do
    jobs = root_jobs_relation(job_class_name:, cursor:).limit(limit + 1).to_a
    build_page(jobs, limit:)
  end
end

#find_job(job_id) ⇒ Hash[String, untyped]?

Note:
  • SolidQueue stores the full ActiveJob serialization in job.arguments
  • We need to extract the actual arguments array for consistency

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

Parameters:

  • (String)

Returns:

  • (Hash[String, untyped], nil) 


157
158
159
160
161
162
163
164
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 157

def find_job(job_id)
  return unless defined?(SolidQueue::Job)

  job = without_query_cache { SolidQueue::Job.find_by(active_job_id: job_id) }
  return if job.nil?

  normalized_job_data(job)
end

#initialize_adapter! ⇒ void

This method returns an undefined value.

: () -> void



22
23
24
25
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 22

def initialize_adapter!
  SolidQueue::Configuration.prepend(SchedulingPatch) if defined?(SolidQueue::Configuration)
  SolidQueue::ClaimedExecution.prepend(ClaimedExecutionPatch) if defined?(SolidQueue::ClaimedExecution)
end

#job_status(job) ⇒ Symbol

: (untyped) -> Symbol

Parameters:

  • (Object)

Returns:

  • (Symbol) 


83
84
85
86
87
88
89
90
91
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 83

def job_status(job)
  without_query_cache do
    return :failed if job.failed?
    return :succeeded if job.finished?
    return :running if job.claimed?

    :pending
  end
end

#normalized_job_data(job) ⇒ Hash[String, untyped]

: (SolidQueue::Job) -> Hash[String, untyped]

Parameters:

  • (SolidQueue::Job)

Returns:

  • (Hash[String, untyped]) 


235
236
237
238
239
240
241
242
243
244
245
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 235

def normalized_job_data(job)
  args = job.arguments
  {
    "job_id" => job.active_job_id,
    "class_name" => job.class_name,
    "queue_name" => job.queue_name,
    "arguments" => args.is_a?(Hash) ? args["arguments"] : args,
    "job_workflow_context" => args.is_a?(Hash) ? args["job_workflow_context"] : nil,
    "status" => job_status(job)
  }
end

#pause_queue(queue_name) ⇒ Boolean

: (String) -> bool

Parameters:

  • (String)

Returns:

  • (Boolean) 


99
100
101
102
103
104
105
106
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 99

def pause_queue(queue_name)
  return false unless defined?(SolidQueue::Queue)

  SolidQueue::Queue.find_by_name(queue_name).pause
  true
rescue ActiveRecord::RecordNotUnique
  true
end

#paused_queues ⇒ Array[String]

: () -> Array

Returns:

  • (Array[String]) 


124
125
126
127
128
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 124

def paused_queues
  return [] unless defined?(SolidQueue::Pause)

  SolidQueue::Pause.pluck(:queue_name)
end

#persist_job_context(job) ⇒ void

Note:
  • Persists the job's updated context (including task outputs) back to the SolidQueue job record after execution completes. Without this, outputs computed during job execution would be lost because SolidQueue does not re-serialize job arguments after perform.

This method returns an undefined value.

: (_JobInterface) -> void

Parameters:

  • (_JobInterface) 


210
211
212
213
214
215
216
217
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 210

def persist_job_context(job)
  return unless defined?(SolidQueue::Job)

  solid_queue_job = SolidQueue::Job.find_by(active_job_id: job.job_id)
  return if solid_queue_job.nil?

  solid_queue_job.update!(arguments: job.serialize.deep_stringify_keys)
end

#queue_latency(queue_name) ⇒ Integer?

: (String) -> Integer?

Parameters:

  • (String)

Returns:

  • (Integer, nil) 


131
132
133
134
135
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 131

def queue_latency(queue_name)
  return nil unless defined?(SolidQueue::Queue)

  SolidQueue::Queue.find_by_name(queue_name).latency
end

#queue_paused?(queue_name) ⇒ Boolean

: (String) -> bool

Parameters:

  • (String)

Returns:

  • (Boolean) 


117
118
119
120
121
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 117

def queue_paused?(queue_name)
  return false unless defined?(SolidQueue::Queue)

  SolidQueue::Queue.find_by_name(queue_name).paused?
end

#queue_size(queue_name) ⇒ Integer

: (String) -> Integer

Parameters:

  • (String)

Returns:

  • (Integer) 


138
139
140
141
142
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 138

def queue_size(queue_name)
  return 0 unless defined?(SolidQueue::Queue)

  SolidQueue::Queue.find_by_name(queue_name).size
end

#reschedule_job(job, wait) ⇒ Boolean

: (_JobInterface, Numeric) -> bool

Parameters:

  • (_JobInterface)
  • (Numeric)

Returns:

  • (Boolean) 


192
193
194
195
196
197
198
199
200
201
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 192

def reschedule_job(job, wait)
  return false unless defined?(SolidQueue::Job)

  solid_queue_job = without_query_cache { SolidQueue::Job.find_by(active_job_id: job.job_id) }
  return false unless solid_queue_job&.claimed?

  reschedule_solid_queue_job(solid_queue_job, job, wait)
rescue ActiveRecord::RecordNotFound
  false
end

#reschedule_solid_queue_job(solid_queue_job, active_job, wait) ⇒ Boolean

: (SolidQueue::Job, _JobInterface, Numeric) -> bool

Parameters:

  • (SolidQueue::Job)
  • (_JobInterface)
  • (Numeric)

Returns:

  • (Boolean) 


263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 263

def reschedule_solid_queue_job(solid_queue_job, active_job, wait)
  solid_queue_job.with_lock do
    solid_queue_job.claimed_execution&.destroy!
    solid_queue_job.update!(
      scheduled_at: wait.seconds.from_now,
      arguments: active_job.serialize.deep_stringify_keys
    )
    solid_queue_job.prepare_for_execution
  end
  true
end

#resume_queue(queue_name) ⇒ Boolean

: (String) -> bool

Parameters:

  • (String)

Returns:

  • (Boolean) 


109
110
111
112
113
114
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 109

def resume_queue(queue_name)
  return false unless defined?(SolidQueue::Queue)

  SolidQueue::Queue.find_by_name(queue_name).resume
  true
end

#root_jobs_relation(job_class_name:, cursor:) ⇒ Object

: (job_class_name: String, cursor: String?) -> untyped

Parameters:

  • job_class_name: (String)
  • cursor: (String, nil)

Returns:

  • (Object) 


248
249
250
251
252
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 248

def root_jobs_relation(job_class_name:, cursor:)
  relation = SolidQueue::Job.where(class_name: job_class_name)
  relation = relation.where("id < ?", cursor.to_i) unless cursor.nil?
  relation.order(id: :desc)
end

#semaphore_available? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


28
29
30
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 28

def semaphore_available?
  defined?(SolidQueue::Semaphore) ? true : false
end

#semaphore_signal(semaphore) ⇒ Boolean

Note:
  • Lifecycle management: The adapter is responsible for removing the hook from SolidQueue::Worker.lifecycle_hooks before calling signal. The hook must be deleted from the registry and the global lifecycle_hooks to prevent redundant signal calls after the semaphore has already been signaled.
  • Hook deletion order: The hook is deleted before calling signal to ensure the hook lambda is no longer invoked even if the signal triggers a worker stop.

: (Semaphore) -> bool

Parameters:

Returns:

  • (Boolean) 


63
64
65
66
67
68
69
70
71
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 63

def semaphore_signal(semaphore)
  return true unless semaphore_available?
  return true unless semaphore_registry.key?(semaphore)

  hook = semaphore_registry[semaphore]
  SolidQueue::Worker.lifecycle_hooks[:stop].delete(hook)
  semaphore_registry.delete(semaphore)
  SolidQueue::Semaphore.signal(semaphore)
end

#semaphore_wait(semaphore) ⇒ Boolean

Note:
  • Thread safety: @semaphore_registry is a non-thread-safe Hash. In multi-threaded workers, concurrent calls to semaphore_wait or semaphore_signal may cause race conditions. Mitigation: SolidQueue workers typically run in single-threaded Fiber mode; verify worker configuration does not enable raw multithreading.
  • Double-wait behavior: If semaphore_wait is called twice for the same Semaphore (e.g., due to retry or requeue), the second call returns false and does not re-register the hook. This is a fail-fast contract: the semaphore is already being waited and will signal the registered hook.

: (Semaphore) -> bool

Parameters:

Returns:

  • (Boolean) 


43
44
45
46
47
48
49
50
51
52
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 43

def semaphore_wait(semaphore)
  return true unless semaphore_available?
  return false if semaphore_registry.key?(semaphore)
  return false unless SolidQueue::Semaphore.wait(semaphore)

  hook = ->(_) { SolidQueue::Semaphore.signal(semaphore) }
  semaphore_registry[semaphore] = hook
  SolidQueue::Worker.on_stop(&hook)
  true
end

#supports_concurrency_limits? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


94
95
96
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 94

def supports_concurrency_limits?
  defined?(SolidQueue) ? true : false
end

#without_query_cache ⇒ void

Note:
  • Bypasses ActiveRecord query cache for the given block.
  • When running under SolidQueue's executor, SELECT queries are cached for the entire job execution. Polling queries must bypass this cache to observe status changes made by other threads/processes.

This method returns an undefined value.

: [T] () { () -> T } -> T



230
231
232
 
# File 'lib/job_workflow/queue_adapters/solid_queue_adapter.rb', line 230

def without_query_cache(&)
  defined?(SolidQueue::Job) ? SolidQueue::Job.uncached(&) : yield
end