Class: DBOperations::HelperWorker

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

Constant Summary collapse

MAIN_KLASS = 
DBOperations::MainWorker

Instance Method Summary collapse

Methods included from HasHelpers::DefaultWorker

included

Instance Method Details

#determine_queue(queue_type) ⇒ Object

Helper method to determine queue type, name and lock key



71
72
73
74
75
76
77
 
# File 'app/workers/has_helpers/db_operations/helper_worker.rb', line 71

def determine_queue(queue_type)
  if queue_type == "fast"
    [MAIN_KLASS::FAST_WORKER_LOCK_KEY, MAIN_KLASS::FAST_QUEUE, true]
  else
    [MAIN_KLASS::SLOW_WORKER_LOCK_KEY, MAIN_KLASS::SLOW_QUEUE, false]
  end
end

#execute_query(job, is_fast) ⇒ Object

Helper method to execute the query, we establish a timeout for fast queues for slow queues we just execute the query without timeout, since we don't know how much they might take The idea is that slow queries should be a rare case and not a common scenario, so we let them be until someone fix them



83
84
85
86
87
 
# File 'app/workers/has_helpers/db_operations/helper_worker.rb', line 83

def execute_query(job, is_fast)
  ::PsqlHq::TimeoutRunner.within(timeout: is_fast ? MAIN_KLASS.config.helper_worker_query_timeout_fast : MAIN_KLASS.config.helper_worker_query_timeout_slow) do |connection|
    job.execute(connection: connection)
  end
end

#log_slow_query(job) ⇒ Object

Log slow query details for later analysis



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

def log_slow_query(job)
  time = Time.zone.now

  # TODO fix this when this is implemented in agencies
  # ex: agencies has its own stat class and field names are different
  stat = ::HasHelpers::Stat.new
  stat.name = job.relation_name
  stat.maintype = "pending_db_ops"
  stat.subtype = "slow"
  stat.data = { id: job.id, query: job.sql, foreign_key: job.foreign_key_value }
  stat.created_at = time
  stat.updated_at = time
  stat.save!
end

#mark_as_slow(job) ⇒ Object

Mark the query as slow



90
91
92
93
94
95
96
 
# File 'app/workers/has_helpers/db_operations/helper_worker.rb', line 90

def mark_as_slow(job)
  remove_job_from_queue(MAIN_KLASS::FAST_QUEUE, job.id)
  job.defer_processing # defer processing so next time is going to be enqueued in the slow queue
  job.is_slow = true
  job.save
  log_slow_query(job)
end

#perform(queue_type) ⇒ Object

queue_type is mandatory and could be one of the following: "fast" runs the worker for the fast queries queue "slow" runs the worker for the slow queries queue



14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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
60
61
62
63
64
65
66
67
68
 
# File 'app/workers/has_helpers/db_operations/helper_worker.rb', line 14

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

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

  # Use appropriate lock and queue names depending on the queue type
  lock_key, queue_name, is_fast = determine_queue(queue_type)

  # Check if another worker is running for this queue; if so, exit early
  return if redis.get(lock_key) == "true"

  # Set a lock in Redis to indicate that this worker is running, with an expiration time
  redis.set(lock_key, "true", ex: MAIN_KLASS.config.helper_worker_lock_expiration)

  # Start renewing the lock in a separate thread (for long-running queries)
  lock_renewal_thread = Thread.new { renew_lock_periodically(lock_key) }

  begin
    # Retrieve and remove a batch of jobs from Redis in an atomic operation
    ids = redis.zpopmin(queue_name, MAIN_KLASS.config.helper_worker_batch_size).map { |id, _score| id }

    return if ids.empty? # If no jobs are left in Redis, exit

    triggered_slow_query_worker = false

    ids.each do |id|
      job = HasHelpers::PendingDBOperation.find_by(id: id)
      begin
        # Process the query (with timeout for fast queries)
        execute_query(job, is_fast) if job
      rescue ::PsqlHq::StatementTimeout
        if is_fast
          # If a query timeout occurs, move it to the slow queue
          mark_as_slow(job)

          # Trigger the slow worker if needed and only once for the whole batch
          if redis.zcard(MAIN_KLASS::SLOW_QUEUE) > 0 && !triggered_slow_query_worker
            ::HasHelpers::DBOperations::HelperWorker.perform_later("slow")
            triggered_slow_query_worker = true
          end
        end
      end
    end

    # Re-enqueue this worker for fast or slow queue if more jobs remain
    ::HasHelpers::DBOperations::HelperWorker.perform_later(queue_type) if redis.zcard(queue_name) > 0
  ensure
    # Stop lock renewal and clear the lock in Redis when the worker finishes
    lock_renewal_thread.kill
    redis.del(lock_key)
  end
end

#redis ⇒ Object 



127
128
129
 
# File 'app/workers/has_helpers/db_operations/helper_worker.rb', line 127

def redis
  RedisConf.current
end

#remove_job_from_queue(queue_name, job_id) ⇒ Object

Remove a job from Redis queue



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

def remove_job_from_queue(queue_name, job_id)
  redis.zrem(queue_name, job_id)
end

#renew_lock_periodically(lock_key) ⇒ Object

Renew the lock periodically (every 4 minutes)



120
121
122
123
124
125
 
# File 'app/workers/has_helpers/db_operations/helper_worker.rb', line 120

def renew_lock_periodically(lock_key)
  loop do
    sleep 1.minute
    redis.expire(lock_key, MAIN_KLASS.config.helper_worker_lock_expiration) if redis.get(lock_key)
  end
end