Class: PatientHttp::Sidekiq::TaskMonitor

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

Overview

Manages inflight request tracking in Redis for crash recovery.

This class maintains a sorted set of request IDs indexed by timestamp and a hash of request payloads. It provides distributed locking for orphan detection and automatic re-enqueuing of requests that were 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

INFLIGHT_INDEX_KEY =

Redis key prefixes

"sidekiq:patient_http:inflight_index"
INFLIGHT_JOBS_KEY = 
"sidekiq:patient_http:inflight_jobs"
INFLIGHT_DETAILS_KEY = 
"sidekiq:patient_http:inflight_details"
INFLIGHT_DETAILS_INDEX_KEY = 
"sidekiq:patient_http:inflight_details_index"
PROCESS_SET_KEY = 
"sidekiq:patient_http:processes"
GC_LOCK_KEY = 
"sidekiq:patient_http:gc_lock"
GC_LAST_RUN_KEY = 
"sidekiq:patient_http:gc_last_run"
REMOVE_IF_ORPHANED_SCRIPT =

Lua script for atomic orphan removal of a batch of request ids. For each id, checks that the task is still orphaned (timestamp < threshold) and removes it atomically, so a heartbeat cannot update the timestamp between the check and the removal. Ids that are no longer orphaned are skipped.

KEYS = index key (sorted set) KEYS = jobs key (hash) KEYS = details key (hash) KEYS = details index key (sorted set) ARGV = threshold_ms ARGV = request_ids

Every removed id is returned, even when the jobs hash no longer holds its payload, so the caller can fall back to the payload it read before the script ran instead of losing the request.

Returns: flat array of [request_id, job_payload, request_id, job_payload, ...] where job_payload is nil when the hash entry was already gone

<<~LUA
  local index_key = KEYS[1]
  local jobs_key = KEYS[2]
  local details_key = KEYS[3]
  local details_index_key = KEYS[4]
  local threshold_ms = tonumber(ARGV[1])
  local removed = {}

  for i = 2, #ARGV do
    local request_id = ARGV[i]
    local current_score = redis.call('ZSCORE', index_key, request_id)
    if current_score and tonumber(current_score) < threshold_ms then
      local job_payload = redis.call('HGET', jobs_key, request_id)
      redis.call('ZREM', index_key, request_id)
      redis.call('HDEL', jobs_key, request_id)
      redis.call('ZREM', details_index_key, request_id)
      redis.call('HDEL', details_key, request_id)
      table.insert(removed, request_id)
      table.insert(removed, job_payload)
    end
  end

  return removed
LUA
REMOVE_IF_ORPHANED_SHA = 
Digest::SHA1.hexdigest(REMOVE_IF_ORPHANED_SCRIPT).freeze
RELEASE_LOCK_SCRIPT =

Lua script for releasing the GC lock only when this process still owns it: a single-round-trip compare-and-delete.

KEYS = lock key ARGV = lock identifier

Returns: 1 if the lock was released, 0 otherwise

<<~LUA
  if redis.call('GET', KEYS[1]) == ARGV[1] then
    return redis.call('DEL', KEYS[1])
  else
    return 0
  end
LUA
RELEASE_LOCK_SHA = 
Digest::SHA1.hexdigest(RELEASE_LOCK_SCRIPT).freeze
ORPHAN_BATCH_SIZE =

Number of orphaned request ids processed per Lua call.

100
MAX_DISPLAY_URL_LENGTH =

Longest URL recorded for the Web UI, so that one enormous URL cannot take a disproportionate amount of memory.

500

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, max_connections: nil, processors: 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. Ignored when a processors source is given, which carries the same information per processor.

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

    callable returning a snapshot of the process's processors as a hash of name => { inflight:, max_capacity: }. The snapshot is published with each heartbeat so the Web UI can report capacity per processor.



383
384
385
386
387
388
389
390
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 383

def initialize(config, max_connections: nil, processors: nil)
  @config = config
  @max_connections_source = max_connections || -> { config.max_connections }
  @processors_source = processors
  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:



99
100
101
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 99

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 registry data. Only allowed in test environment.

Raises:

  • (RuntimeError) —

    if called outside of test environment



276
277
278
279
280
281
282
283
284
285
286
287
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 276

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

  ::Sidekiq.redis do |redis|
    redis.del(
      INFLIGHT_INDEX_KEY, INFLIGHT_JOBS_KEY, INFLIGHT_DETAILS_KEY,
      INFLIGHT_DETAILS_INDEX_KEY, PROCESS_SET_KEY, GC_LOCK_KEY, GC_LAST_RUN_KEY
    )
  end
end

.inflight_count ⇒ Integer

Get the count of inflight requests in Redis.

Returns:

  • (Integer) —

    number of inflight requests



105
106
107
108
109
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 105

def inflight_count
  ::Sidekiq.redis do |redis|
    redis.zcard(INFLIGHT_INDEX_KEY)
  end
end

.inflight_counts_by_process ⇒ Hash

Get all inflight counts across all processes and the number of max connections.

The per-process inflight count comes from the shared inflight index, so it includes requests left behind by processes that have since died. The nested per-processor counts are snapshots each process publishes with its heartbeat, so they only cover processes that are still running and can lag by up to one monitor cycle.

Returns:

  • (Hash) —

    hash of "hostname:pid" => { inflight: Integer, max_capacity: Integer, processors: { String => { inflight: Integer, max_capacity: Integer } } }



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 122

def inflight_counts_by_process
  process_ids = nil
  max_connections = nil
  processor_snapshots = nil
  inflight_task_ids = nil

  ::Sidekiq.redis do |redis|
    process_ids = redis.smembers(PROCESS_SET_KEY)
    return {} if process_ids.empty?

    max_keys = process_ids.map { |pid| max_connections_key_for(pid) }
    processor_keys = process_ids.map { |pid| processors_key_for(pid) }
    values = redis.mget(*max_keys, *processor_keys)
    max_connections = values.first(process_ids.size)
    processor_snapshots = values.last(process_ids.size)

    inflight_task_ids = redis.zrange(INFLIGHT_INDEX_KEY, 0, -1)
  end

  inflight_by_process_id = inflight_task_ids.group_by do |task_id|
    task_id.split("/", 2).first
  end

  result = {}
  stale_process_ids = []

  process_ids.zip(max_connections, processor_snapshots).each do |process_id, max_conn, snapshot|
    if max_conn.nil?
      # Mark for removal if max_conn key doesn't exist (process is gone)
      stale_process_ids << process_id
    else
      host_pid = process_id.split(":", 3).first(2).join(":")
      counts = result[host_pid]
      unless counts
        counts = {inflight: 0, max_capacity: 0, processors: {}}
        result[host_pid] = counts
      end
      counts[:inflight] += inflight_by_process_id[process_id]&.size.to_i
      counts[:max_capacity] += max_conn.to_i
      merge_processor_snapshot(counts[:processors], snapshot)
    end
  end

  # Remove stale process IDs from the set
  unless stale_process_ids.empty?
    ::Sidekiq.redis do |redis|
      redis.srem(PROCESS_SET_KEY, stale_process_ids)
    end
  end

  result
end

.inflight_counts_by_processor(processes = nil) ⇒ Hash

Get the inflight and capacity counts for each named processor across all running processes.

Parameters:

Returns:

  • (Hash) —

    hash of processor name => { inflight: Integer, max_capacity: Integer }



181
182
183
184
185
186
187
188
189
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 181

def inflight_counts_by_processor(processes = nil)
  processes ||= inflight_counts_by_process

  result = {}
  processes.each_value do |data|
    merge_processor_counts(result, data[:processors])
  end
  result.sort.to_h
end

.inflight_details(limit: 50) ⇒ Array<Hash>

Get the details of the requests that have been in flight the longest.

Only requests registered while inflight_details was enabled are reported. A request stays listed while its crash-recovery record exists, so a request left behind by a process that died is listed until the orphan collector re-enqueues it.

Parameters:

  • limit (Integer) (defaults to: 50) —

    maximum number of requests to return

Returns:

  • (Array<Hash>) —

    oldest first, each with :request_id, :process_id, :url, :http_method, :processor, and :age in seconds



201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 201

def inflight_details(limit: 50)
  return [] if limit <= 0

  task_ids = nil
  timestamps = nil
  records = nil

  ::Sidekiq.redis do |redis|
    entries = redis.zrange(INFLIGHT_DETAILS_INDEX_KEY, 0, limit - 1, withscores: true)
    return [] if entries.empty?

    task_ids = entries.map(&:first)
    timestamps = entries.map(&:last)
    records = redis.hmget(INFLIGHT_DETAILS_KEY, *task_ids)
  end

  now = Time.now.to_f
  task_ids.zip(timestamps, records).filter_map do |task_id, timestamp_ms, record|
    details = parse_details(record)
    next unless details

    process_id, request_id = task_id.split("/", 2)
    {
      request_id: request_id,
      process_id: process_id.to_s.split(":", 3).first(2).join(":"),
      url: details["url"],
      http_method: details["method"],
      processor: details["processor"],
      age: (now - timestamp_ms.to_f / 1000.0).round(1)
    }
  end
end

.registered_process_ids ⇒ Array<String>

Get all registered process IDs.

Returns:

  • (Array<String>) —

    list of process identifiers



265
266
267
268
269
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 265

def registered_process_ids
  ::Sidekiq.redis do |redis|
    redis.smembers(PROCESS_SET_KEY)
  end
end

.sanitize_url(url) ⇒ String

Remove the user name, password, query string, and fragment from a URL, keeping the scheme, host, and path. Used unless the configuration names its own sanitizer.

Parameters:

  • url (String) —

    the request URL

Returns:

  • (String) —

    the URL to display



240
241
242
243
244
245
246
247
248
249
250
251
252
253
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 240

def sanitize_url(url)
  uri = URI.parse(url.to_s)
  uri.query = nil
  uri.fragment = nil
  # The password must be cleared before the user, and clearing the user
  # info in one step does nothing.
  uri.password = nil if uri.respond_to?(:password=)
  uri.user = nil if uri.respond_to?(:user=)
  uri.to_s
rescue
  # A URL that cannot be parsed, such as one with a character outside
  # US-ASCII, still must not carry credentials or a query string.
  strip_credentials(url.to_s.split(/[?#]/, 2).first.to_s)
end

.total_max_connections ⇒ Integer

Get the total max connections across all processes

Returns:

  • (Integer) —

    sum of max connections from all active processes



258
259
260
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 258

def total_max_connections
  inflight_counts_by_process.values.sum { |data| data[:max_capacity] }
end

Instance Method Details

#acquire_gc_lock ⇒ Boolean

Try to acquire the distributed garbage collection lock.

Returns:

  • (Boolean) —

    true if lock acquired, false otherwise



553
554
555
556
557
558
559
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 553

def acquire_gc_lock
  PatientHttp::Sidekiq.redis do |redis|
    # Use SET with NX and EX options directly
    # Returns "OK" if successful, nil if key already exists
    !!redis.set(GC_LOCK_KEY, @lock_identifier, nx: true, ex: gc_lock_ttl)
  end
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



609
610
611
612
613
614
615
616
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 609

def cleanup_orphaned_requests(orphan_threshold_seconds, logger)
  threshold_timestamp_ms = calculate_threshold_timestamp(orphan_threshold_seconds)
  orphaned_requests = fetch_orphaned_requests(threshold_timestamp_ms)

  return 0 if orphaned_requests.empty?

  reenqueue_orphaned_jobs(orphaned_requests, threshold_timestamp_ms, logger)
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

Returns:

  • (String) —

    the unique task ID



517
518
519
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 517

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

#gc_needed? ⇒ Boolean

Check if garbage collection should run based on the last run timestamp.

Returns true if the GC_LAST_RUN_KEY doesn't exist in Redis or if enough time has elapsed since the last GC run.

Returns:

  • (Boolean) —

    true if GC should run, false otherwise



580
581
582
583
584
585
586
587
588
589
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 580

def gc_needed?
  last_run = PatientHttp::Sidekiq.redis do |redis|
    redis.get(GC_LAST_RUN_KEY)
  end

  return true if last_run.nil?

  last_run_time = Time.at(last_run.to_f / 1000.0)
  Time.now - last_run_time >= config.heartbeat_interval
end

#heartbeat_timestamp_for(task) ⇒ Integer?

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.

Get the heartbeat timestamp for a task.

Parameters:

  • task (RequestTask) —

    the request task

Returns:

  • (Integer, nil) —

    timestamp in milliseconds, or nil if not registered



496
497
498
499
500
501
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 496

def heartbeat_timestamp_for(task)
  score = PatientHttp::Sidekiq.redis do |redis|
    redis.zscore(INFLIGHT_INDEX_KEY, full_task_id(task.id))
  end
  score&.to_i
end

#ping_process ⇒ void

This method returns an undefined value.

Record the current process's capacity in Redis.

This is used for monitoring purposes. The max connections key doubles as the process's liveness marker: it is refreshed on every heartbeat with a TTL shorter than the process set's, so a member of the set whose key has expired belongs to a process that is gone.



529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 529

def ping_process
  snapshot = @processors_source&.call
  max_connections = if snapshot
    snapshot.values.sum { |counts| counts[:max_capacity].to_i }
  else
    @max_connections_source.call
  end

  PatientHttp::Sidekiq.redis do |redis|
    redis.multi do |transaction|
      transaction.sadd(PROCESS_SET_KEY, @lock_identifier)
      transaction.set(max_connections_key, max_connections)
      transaction.expire(PROCESS_SET_KEY, inflight_ttl)
      transaction.expire(max_connections_key, process_ttl)
      if snapshot
        transaction.set(processors_key, serialize_processor_snapshot(snapshot), ex: process_ttl)
      end
    end
  end
end

#record_gc_run ⇒ void

This method returns an undefined value.

Record the timestamp of the last GC run in Redis.

The timestamp is stored with a TTL slightly longer than the heartbeat interval to coordinate GC execution across multiple processes.



597
598
599
600
601
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 597

def record_gc_run
  PatientHttp::Sidekiq.redis do |redis|
    redis.set(GC_LAST_RUN_KEY, (Time.now.to_f * 1000).floor, ex: gc_last_run_ttl)
  end
end

#register(task, processor_name: nil) ⇒ void

This method returns an undefined value.

Register a request as inflight in Redis.

Parameters:

  • task (RequestTask) —

    the request task to register

  • processor_name (Symbol, String, nil) (defaults to: nil) —

    name of the processor running the request, recorded with the request details



399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 399

def register(task, processor_name: nil)
  timestamp_ms = (Time.now.to_f * 1000).round
  job_payload = JSON.generate(task.task_handler.sidekiq_job)
  task_id = full_task_id(task.id)
  details = request_details(task, processor_name)

  PatientHttp::Sidekiq.redis do |redis|
    redis.multi do |transaction|
      transaction.zadd(INFLIGHT_INDEX_KEY, timestamp_ms, task_id)
      transaction.hset(INFLIGHT_JOBS_KEY, task_id, job_payload)
      transaction.expire(INFLIGHT_INDEX_KEY, inflight_ttl)
      transaction.expire(INFLIGHT_JOBS_KEY, inflight_ttl)
      if details
        transaction.zadd(INFLIGHT_DETAILS_INDEX_KEY, timestamp_ms, task_id)
        transaction.hset(INFLIGHT_DETAILS_KEY, task_id, details)
        transaction.expire(INFLIGHT_DETAILS_INDEX_KEY, inflight_ttl)
        transaction.expire(INFLIGHT_DETAILS_KEY, inflight_ttl)
      end
    end
  end
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 registry.

Parameters:

  • task (RequestTask) —

    the request task

Returns:

  • (Boolean) —

    true if registered, false otherwise



484
485
486
487
488
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 484

def registered?(task)
  PatientHttp::Sidekiq.redis do |redis|
    !redis.zscore(INFLIGHT_INDEX_KEY, full_task_id(task.id)).nil?
  end
end

#registered_task_ids ⇒ Array<String>

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.

Get all registered task IDs for this registry's process.

Returns:

  • (Array<String>) —

    list of full task IDs



507
508
509
510
511
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 507

def registered_task_ids
  PatientHttp::Sidekiq.redis do |redis|
    redis.zrange(INFLIGHT_INDEX_KEY, 0, -1)
  end.select { |id| id.start_with?("#{@lock_identifier}/") }
end

#release_gc_lock ⇒ Boolean

Release the garbage collection lock if held by this process.

Uses a compare-and-delete Lua script so the check and deletion happen atomically in a single round trip.

Returns:

  • (Boolean) —

    true if the lock was released, false otherwise



567
568
569
570
571
572
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 567

def release_gc_lock
  result = PatientHttp::Sidekiq.redis do |redis|
    run_script(redis, RELEASE_LOCK_SCRIPT, RELEASE_LOCK_SHA, [GC_LOCK_KEY], [@lock_identifier])
  end
  result == 1
end

#remove_process ⇒ void

This method returns an undefined value.

Remove this process's entry from the process set.



442
443
444
445
446
447
448
449
450
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 442

def remove_process
  PatientHttp::Sidekiq.redis do |redis|
    redis.pipelined do |pipeline|
      pipeline.srem(PROCESS_SET_KEY, @lock_identifier)
      pipeline.del(max_connections_key)
      pipeline.del(processors_key)
    end
  end
end

#unregister(task) ⇒ void

This method returns an undefined value.

Unregister a request from Redis (called when request completes).

Parameters:

  • task (RequestTask) —

    the request task to unregister



426
427
428
429
430
431
432
433
434
435
436
437
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 426

def unregister(task)
  task_id = full_task_id(task.id)

  PatientHttp::Sidekiq.redis do |redis|
    redis.multi do |transaction|
      transaction.zrem(INFLIGHT_INDEX_KEY, task_id)
      transaction.hdel(INFLIGHT_JOBS_KEY, task_id)
      transaction.zrem(INFLIGHT_DETAILS_INDEX_KEY, task_id)
      transaction.hdel(INFLIGHT_DETAILS_KEY, task_id)
    end
  end
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



457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
 
# File 'lib/patient_http/sidekiq/task_monitor.rb', line 457

def update_heartbeats(task_ids)
  return if task_ids.empty?

  timestamp_ms = (Time.now.to_f * 1000).round

  PatientHttp::Sidekiq.redis do |redis|
    redis.pipelined do |pipeline|
      task_ids.each do |task_id|
        pipeline.call("ZADD", INFLIGHT_INDEX_KEY, "XX", timestamp_ms, full_task_id(task_id))
      end
      # Keep the inflight keys alive while requests are still in flight;
      # otherwise they only get their TTL refreshed when new requests
      # are registered.
      pipeline.call("EXPIRE", INFLIGHT_INDEX_KEY, inflight_ttl)
      pipeline.call("EXPIRE", INFLIGHT_JOBS_KEY, inflight_ttl)
      pipeline.call("EXPIRE", INFLIGHT_DETAILS_INDEX_KEY, inflight_ttl)
      pipeline.call("EXPIRE", INFLIGHT_DETAILS_KEY, inflight_ttl)
    end
  end
end