Class: PatientHttp::SolidQueue::TaskMonitor

Inherits:
Object
  • Object
show all
Defined in:
lib/patient_http/solid_queue/task_monitor.rb

Overview

Manages inflight request tracking in the database for crash recovery.

This class maintains Active Record records for each in-flight request. It provides distributed locking for orphan detection and automatic re-enqueueing of requests interrupted by process crashes.

Task ID format: "hostname:pid:hex/request-uuid"

  • hostname: sanitized hostname (colons and slashes replaced with dashes)
  • pid: process ID
  • hex: 8-character random hex for uniqueness
  • request-uuid: unique identifier for the request

Constant Summary collapse

GC_LOCK_NAME = 
"gc"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, max_connections: nil) ⇒ TaskMonitor

Returns a new instance of TaskMonitor.

Parameters:

  • config (Configuration) —

    the configuration object

  • max_connections (#call, nil) (defaults to: nil) —

    callable returning the process's total configured max connections; defaults to the configuration's value. With named processors the module passes a sum across all processors.



26
27
28
29
30
31
32
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 26

def initialize(config, max_connections: nil)
  @config = config
  @max_connections_source = max_connections || -> { config.max_connections }
  hostname = ::Socket.gethostname.force_encoding("UTF-8").tr(":/", "-")
  pid = ::Process.pid
  @lock_identifier = "#{hostname}:#{pid}:#{SecureRandom.hex(8)}".freeze
end

Instance Attribute Details

#config ⇒ Configuration (readonly)

Returns the configuration object.

Returns:



20
21
22
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 20

def config
  @config
end

Class Method Details

.clear_all! ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Clear all records. Only allowed in test environment.

Raises:

  • (RuntimeError) —

    if called outside of test environment



249
250
251
252
253
254
255
256
257
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 249

def self.clear_all!
  unless PatientHttp.testing?
    raise "clear_all! is only allowed in test environment"
  end

  InflightRequest.delete_all
  ProcessRegistration.delete_all
  GcLock.delete_all
end

Instance Method Details

#acquire_gc_lock ⇒ Boolean

Try to acquire the distributed garbage collection lock.

Uses a single semaphore row and pessimistic locking to ensure only one process can claim the lock at a time. Returns false if another process holds a non-expired lock, or if GC was run recently (within heartbeat_interval).

Returns:

  • (Boolean) —

    true if lock acquired, false otherwise



153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 153

def acquire_gc_lock
  now = Time.current
  expires_at = now + gc_lock_ttl.seconds
  acquired = false

  ensure_gc_lock_row!

  GcLock.transaction do
    lock = GcLock.lock.find_by!(lock_name: GC_LOCK_NAME)

    recent_gc = lock.last_gc_at && lock.last_gc_at > (now - @config.heartbeat_interval)
    next if recent_gc

    lock_held = lock.lock_holder.present? && lock.expires_at.present? && lock.expires_at > now
    next if lock_held

    lock.update!(
      lock_holder: @lock_identifier,
      acquired_at: now,
      expires_at: expires_at
    )
    acquired = true
  end

  acquired
rescue => e
  @config.logger&.error("[PatientHttp::SolidQueue] Failed to acquire GC lock: #{e.message}")
  raise if PatientHttp.testing?
  false
end

#cleanup_orphaned_requests(orphan_threshold_seconds, logger) ⇒ Integer

Find and re-enqueue orphaned requests.

Parameters:

  • orphan_threshold_seconds (Numeric) —

    age threshold for considering a request orphaned

  • logger (Logger) —

    logger for output

Returns:

  • (Integer) —

    number of orphaned requests re-enqueued



200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 200

def cleanup_orphaned_requests(orphan_threshold_seconds, logger)
  threshold = Time.current - orphan_threshold_seconds.seconds

  prune_stale_process_registrations(threshold)

  # Get process IDs with a recent heartbeat
  active_process_ids = ProcessRegistration.where("last_seen_at >= ?", threshold).pluck(:process_id)

  # Find stale requests from processes not in the active set
  orphaned = InflightRequest
    .where("heartbeat_at < ?", threshold)
    .where.not(process_id: active_process_ids)
    .to_a

  return 0 if orphaned.empty?

  reenqueued_count = 0

  orphaned.each do |record|
    reenqueued_count += 1 if reenqueue_orphaned_record(record, threshold, logger)
  end

  reenqueued_count
end

#full_task_id(task_id) ⇒ String

Build unique task ID for a request task that includes process identifier.

Parameters:

  • task_id (String) —

    the request task ID

Returns:

  • (String) —

    the unique task ID



229
230
231
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 229

def full_task_id(task_id)
  "#{@lock_identifier}/#{task_id}"
end

#ping_process ⇒ void

This method returns an undefined value.

Record or refresh this process's registration.



119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 119

def ping_process
  max_connections = @max_connections_source.call

  with_connection do
    ProcessRegistration.upsert(
      {process_id: @lock_identifier, max_connections: max_connections, last_seen_at: Time.current},
      unique_by: upsert_unique_by(:process_id)
    )
  end
rescue => e
  @config.logger&.error("[PatientHttp::SolidQueue] Failed to ping process: #{e.message}")
  raise if PatientHttp.testing?
end

#register(task) ⇒ void

This method returns an undefined value.

Register a request as inflight in the database.

Runs on the caller's thread via the request_enqueued observer event. Errors propagate so a task is never accepted without a durable record: the processor rejects the task and the enqueue raises to the caller. They are wrapped in RegistrationError so the job retries instead of failing outright, since the usual cause is a transient database issue.

Parameters:

  • task (PatientHttp::RequestTask) —

    the request task to register

Raises:



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 45

def register(task)
  job_payload = task.task_handler.active_job_data.to_json
  task_id = full_task_id(task.id)
  now = Time.current

  with_connection do
    InflightRequest.create!(
      task_id: task_id,
      process_id: @lock_identifier,
      job_payload: job_payload,
      heartbeat_at: now,
      created_at: now
    )
  end
rescue => e
  @config.logger&.error("[PatientHttp::SolidQueue] Failed to register task #{task_id}: #{e.class} - #{e.message}")
  raise RegistrationError.new("Failed to register task #{task_id}: #{e.class} - #{e.message}")
end

#registered?(task) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Check if a task is registered in the inflight table.

Parameters:

  • task (PatientHttp::RequestTask) —

    the request task

Returns:

  • (Boolean) 


238
239
240
241
242
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 238

def registered?(task)
  with_connection do
    InflightRequest.where(task_id: full_task_id(task.id)).exists?
  end
end

#release(task) ⇒ void

This method returns an undefined value.

Release a request from this process so the orphan collector re-enqueues it on its next pass. Used when a result could not be delivered: the request is no longer tracked here, so its record must not keep looking like it belongs to a live process. Orphan collection skips records whose process is still registered, which would otherwise strand the request until this process exits.

Parameters:

  • task (PatientHttp::RequestTask) —

    the request task to release



87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 87

def release(task)
  task_id = full_task_id(task.id)
  with_connection do
    InflightRequest.where(task_id: task_id).update_all(
      process_id: released_process_id,
      heartbeat_at: Time.at(0).utc
    )
  end
rescue => e
  @config.logger&.error("[PatientHttp::SolidQueue] Failed to release task #{task_id}: #{e.message}")
  raise if PatientHttp.testing?
end

#release_gc_lock ⇒ void

This method returns an undefined value.

Release the garbage collection lock if held by this process, and record last_gc_at.



187
188
189
190
191
192
193
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 187

def release_gc_lock
  GcLock.where(lock_name: GC_LOCK_NAME, lock_holder: @lock_identifier)
    .update_all(last_gc_at: Time.current, lock_holder: nil, acquired_at: nil, expires_at: nil)
rescue => e
  @config.logger&.error("[PatientHttp::SolidQueue] Failed to release GC lock: #{e.message}")
  raise if PatientHttp.testing?
end

#remove_process ⇒ void

This method returns an undefined value.

Remove this process's registration.



136
137
138
139
140
141
142
143
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 136

def remove_process
  with_connection do
    ProcessRegistration.where(process_id: @lock_identifier).delete_all
  end
rescue => e
  @config.logger&.error("[PatientHttp::SolidQueue] Failed to remove process: #{e.message}")
  raise if PatientHttp.testing?
end

#unregister(task) ⇒ void

This method returns an undefined value.

Unregister a request from the database (called when request completes).

Parameters:

  • task (PatientHttp::RequestTask) —

    the request task to unregister



68
69
70
71
72
73
74
75
76
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 68

def unregister(task)
  task_id = full_task_id(task.id)
  with_connection do
    InflightRequest.where(task_id: task_id).delete_all
  end
rescue => e
  @config.logger&.error("[PatientHttp::SolidQueue] Failed to unregister task #{task_id}: #{e.message}")
  raise if PatientHttp.testing?
end

#update_heartbeats(task_ids) ⇒ void

This method returns an undefined value.

Update heartbeat timestamps for multiple requests in a single operation.

Parameters:

  • task_ids (Array<String>) —

    the request IDs to update



104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/patient_http/solid_queue/task_monitor.rb', line 104

def update_heartbeats(task_ids)
  return if task_ids.empty?

  full_ids = task_ids.map { |id| full_task_id(id) }
  with_connection do
    InflightRequest.where(task_id: full_ids).update_all(heartbeat_at: Time.current)
  end
rescue => e
  @config.logger&.error("[PatientHttp::SolidQueue] Failed to update heartbeats: #{e.message}")
  raise if PatientHttp.testing?
end