Class: DBOperations::MainWorker

Inherits:
HasUtils::GoodJob::BaseWorker
  • Object
show all
Includes:
HasHelpers::DefaultWorker, HasHelpers::RedisConfigurable
Defined in:
app/workers/has_helpers/db_operations/main_worker.rb

Constant Summary collapse

FAST_QUEUE =

The queue used for fast/usual jobs

"pending_db_ops_fast"
SLOW_QUEUE =

The queue used for jobs that were marked as slow

"pending_db_ops_slow"
FAST_WORKER_LOCK_KEY =

The lock key for the worker that executes jobs in the fast queue

"pending_db_ops_fast_worker"
SLOW_WORKER_LOCK_KEY =

The lock key for the worker that executes jobs in the slow queue

"pending_db_ops_slow_worker"
MAIN_WORKER_LOCK_KEY =

The lock key for the main worker

"pending_db_ops_main_worker"

Instance Method Summary collapse

Methods included from HasHelpers::RedisConfigurable

included

Methods included from HasHelpers::DefaultWorker

included

Instance Method Details

#enqueue_all_pending_jobs ⇒ Object 



115
116
117
118
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 115

def enqueue_all_pending_jobs
  enqueue_pending_jobs(nil) # enqueue fast queries
  enqueue_pending_jobs(true) # enqueue slow queries
end

#enqueue_job_in_redis(job, queue_name) ⇒ Object

Helper method to enqueue a job into Redis



148
149
150
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 148

def enqueue_job_in_redis(job, queue_name)
  redis.zadd(queue_name, job.created_at.to_f, job.id)
end

#enqueue_jobs(jobs) ⇒ Object 



135
136
137
138
139
140
141
142
143
144
145
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 135

def enqueue_jobs(jobs)
  jobs.each do |job|
    break if ::HasHelpers::DBOperations::Operations.paused?
    if job.is_slow
      enqueue_job_in_redis(job, SLOW_QUEUE) # Enqueue slow jobs directly into the slow queue
    else
      enqueue_job_in_redis(job, FAST_QUEUE) # Enqueue regular jobs into the fast queue
    end
    job.update!(enqueued_at: Time.zone.now)
  end
end

#enqueue_pending_jobs(is_slow) ⇒ Object

Enqueue new jobs that have not been enqueued yet



121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 121

def enqueue_pending_jobs(is_slow)
  queue_name = is_slow ? SLOW_QUEUE : FAST_QUEUE
  current_queue_size = redis.zcard(queue_name)
  return if current_queue_size >= config.max_queue_size

  current_limit = [config.enqueue_batch_size, config.max_queue_size - current_queue_size].min

  pending_jobs_sql = HasHelpers::PendingDBOperation.where(enqueued_at: nil).where(is_slow: is_slow).order(:created_at).limit(current_limit).to_sql
  HasHelpers::PendingDBOperation.uncached do
    pending_jobs = HasHelpers::PendingDBOperation.find_by_sql(pending_jobs_sql) # The reason behind doing this is to bypass ActiveRecord cache, I tried with just `uncached` but Rails was still caching the result
    enqueue_jobs(pending_jobs)
  end
end

#lock_main_worker ⇒ Object 



65
66
67
68
69
70
71
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 65

def lock_main_worker
  if redis.get(MAIN_WORKER_LOCK_KEY) == "true"
    redis.expire(MAIN_WORKER_LOCK_KEY, 1.minute)
  else
    redis.set(MAIN_WORKER_LOCK_KEY, "true", ex: 1.minute)
  end
end

#main_process ⇒ Object 



77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 77

def main_process
  # Recover jobs stuck in both fast and slow queues
  recover_all_stuck_jobs

  # Enqueue pending jobs in both fast and slow queues
  enqueue_all_pending_jobs

  # If there are jobs in the fast queue and no fast worker is running, trigger the fast worker
  if redis.zcard(FAST_QUEUE) > 0 && !worker_running?(FAST_WORKER_LOCK_KEY)
    ::HasHelpers::DBOperations::HelperWorker.perform_later("fast")
  end

  # If there are jobs in the slow queue and no slow worker is running, trigger the slow worker
  if redis.zcard(SLOW_QUEUE) > 0 && !worker_running?(SLOW_WORKER_LOCK_KEY)
    ::HasHelpers::DBOperations::HelperWorker.perform_later("slow")
  end
end

#main_worker_locked? ⇒ Boolean

Returns:

  • (Boolean) 


61
62
63
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 61

def main_worker_locked?
  redis.get(MAIN_WORKER_LOCK_KEY) == "true"
end

#perform ⇒ Object 



35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 35

def perform
  return if HasHelpers::Feature.active?(:old_pending_db_ops)

  if ::HasHelpers::DBOperations::Operations.paused?
    ::Rails.logger.info "DBOperations::MainWorker skip because paused"
    return
  end

  begin
    return if main_worker_locked?
    started_at = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)

    loop do
      lock_main_worker
      main_process

      ended_at = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
      sleep(config.main_worker_sleep.seconds)
      break if (ended_at - started_at) > config.main_worker_lifespan.seconds
      break if ::HasHelpers::DBOperations::Operations.paused?
    end
  ensure
    release_main_worker_lock
  end
end

#recover_all_stuck_jobs ⇒ Object 



95
96
97
98
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 95

def recover_all_stuck_jobs
  recover_stuck_jobs(nil) # recover fast stuck queries
  recover_stuck_jobs(true) # recover slow stuck queries
end

#recover_stuck_jobs(is_slow) ⇒ Object

Recover stuck jobs in the queue and re-enqueue them



101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 101

def recover_stuck_jobs(is_slow)
  queue_name = is_slow ? SLOW_QUEUE : FAST_QUEUE
  current_queue_size = redis.zcard(queue_name)
  return if current_queue_size >= config.max_queue_size

  current_limit = [config.enqueue_batch_size, config.max_queue_size - current_queue_size].min

  stuck_jobs_sql = HasHelpers::PendingDBOperation.where(HasHelpers::PendingDBOperation.arel_table[:enqueued_at].lt(Time.zone.now - config.max_enqueue_time.seconds)).where.not(enqueued_at: nil).where(is_slow: is_slow).limit(current_limit).to_sql
  HasHelpers::PendingDBOperation.uncached do
    stuck_jobs = HasHelpers::PendingDBOperation.find_by_sql(stuck_jobs_sql) # The reason behind doing this is to bypass ActiveRecord cache, I tried with just `uncached` but Rails was still caching the result
    enqueue_jobs(stuck_jobs)
  end
end

#redis ⇒ Object 



158
159
160
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 158

def redis
  RedisConf.current
end

#release_main_worker_lock ⇒ Object 



73
74
75
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 73

def release_main_worker_lock
  redis.del(MAIN_WORKER_LOCK_KEY)
end

#worker_running?(lock_key) ⇒ Boolean

Check if the worker for the fast or slow queue is already running by looking at a Redis lock/flag with expiration

Returns:

  • (Boolean) 


153
154
155
156
 
# File 'app/workers/has_helpers/db_operations/main_worker.rb', line 153

def worker_running?(lock_key)
  lock_ttl = redis.ttl(lock_key)
  lock_ttl > 0 # The lock is active if TTL (Time to Live) is positive
end