Module: SidekiqSolidFetch

Defined in:
lib/sidekiq_solid_fetch.rb,
lib/sidekiq_solid_fetch/fetcher.rb,
lib/sidekiq_solid_fetch/version.rb,
lib/sidekiq_solid_fetch/throttled.rb,
lib/sidekiq_solid_fetch/unit_of_work.rb

Defined Under Namespace

Modules: Throttled Classes: Fetcher, UnitOfWork

Constant Summary collapse

PROCESSING_QUEUE_PREFIX = 
"sidekiq_solid_fetch:processing"
REGISTRY_KEY =

Registry of every live processing queue. A Redis hash where each field is a processing queue key and each value is a JSON blob with the original queue, the worker identity and the registration timestamp. Recovery reads this registry instead of parsing key names, which would be ambiguous because both queue names and identities can contain colons.

"sidekiq_solid_fetch:registry"
RECOVERY_LOCK_KEY = 
"sidekiq_solid_fetch:recovery_lock"
GRACE_PERIOD =

Entries younger than this are never considered orphaned, so a freshly booted worker whose first heartbeat hasn't landed yet can't have its in-flight jobs stolen by another process running recovery.

60
RECOVERY_INTERVAL =

How often each process attempts orphan recovery after boot. Runs are deduplicated across processes with RECOVERY_LOCK_KEY.

300
VERSION = 
"0.2.0"

Class Method Summary collapse

Class Method Details

.acquire_recovery_lock(config) ⇒ Object 



103
104
105
 
# File 'lib/sidekiq_solid_fetch.rb', line 103

def self.acquire_recovery_lock(config)
  config.redis { |conn| conn.set(RECOVERY_LOCK_KEY, "1", "NX", "EX", RECOVERY_INTERVAL) }
end

.enable!(config) ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/sidekiq_solid_fetch.rb', line 28

def self.enable!(config)
  config[:fetch_class] = SidekiqSolidFetch::Fetcher

  config.on(:startup) do
    config.logger.info("SidekiqSolidFetch enabled")
    requeue_orphaned_jobs(config)
    @recovery_thread = start_recovery_thread(config)
  end

  config.on(:shutdown) do
    @recovery_thread&.kill
    @recovery_thread = nil
  end
end

.parse_registry_entry(raw_entry) ⇒ Object 



107
108
109
110
111
112
113
 
# File 'lib/sidekiq_solid_fetch.rb', line 107

def self.parse_registry_entry(raw_entry)
  entry = JSON.parse(raw_entry)
  return nil unless entry.is_a?(Hash) && entry["queue"] && entry["identity"]
  entry
rescue JSON::ParserError
  nil
end

.processing_queue_name(queue, identity, capsule_name) ⇒ Object 



77
78
79
 
# File 'lib/sidekiq_solid_fetch.rb', line 77

def self.processing_queue_name(queue, identity, capsule_name)
  "#{PROCESSING_QUEUE_PREFIX}:#{capsule_name}:#{identity}:#{queue}"
end

.register_processing_queue(conn, processing_queue, queue, identity) ⇒ Object 



81
82
83
84
 
# File 'lib/sidekiq_solid_fetch.rb', line 81

def self.register_processing_queue(conn, processing_queue, queue, identity)
  entry = {"queue" => queue, "identity" => identity, "registered_at" => Time.now.to_i}
  conn.hset(REGISTRY_KEY, processing_queue, JSON.generate(entry))
end

.requeue_orphaned_jobs(config) ⇒ Object

Requeue jobs from processing queues whose worker is dead. A worker is considered dead when its heartbeat key (which Sidekiq maintains with a TTL) no longer exists. Membership in the processes set is not used because entries for crashed workers linger there until Sidekiq's rate-limited cleanup runs.



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/sidekiq_solid_fetch.rb', line 48

def self.requeue_orphaned_jobs(config)
  logger = config.logger
  logger.info("SidekiqSolidFetch: Checking for orphaned jobs from dead workers")

  count = 0
  now = Time.now.to_i
  config.redis do |conn|
    conn.hgetall(REGISTRY_KEY).each do |processing_queue, raw_entry|
      entry = parse_registry_entry(raw_entry)
      unless entry
        conn.hdel(REGISTRY_KEY, processing_queue)
        next
      end

      next if now - entry["registered_at"].to_i < GRACE_PERIOD
      next if conn.exists(entry["identity"]) > 0

      queue = entry["queue"]
      while conn.lmove(processing_queue, queue, "LEFT", "RIGHT")
        count += 1
        logger.info { "SidekiqSolidFetch: Moved orphaned job from #{processing_queue} back to #{queue}" }
      end
      conn.hdel(REGISTRY_KEY, processing_queue)
    end
  end

  logger.info("SidekiqSolidFetch: Re-queued #{count} orphaned jobs")
end

.start_recovery_thread(config) ⇒ Object 



90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/sidekiq_solid_fetch.rb', line 90

def self.start_recovery_thread(config)
  Thread.new do
    loop do
      sleep(RECOVERY_INTERVAL)
      begin
        requeue_orphaned_jobs(config) if acquire_recovery_lock(config)
      rescue => ex
        config.logger.warn("SidekiqSolidFetch: Orphan recovery failed: #{ex.message}")
      end
    end
  end
end

.unregister_processing_queues(conn, processing_queues) ⇒ Object 



86
87
88
 
# File 'lib/sidekiq_solid_fetch.rb', line 86

def self.unregister_processing_queues(conn, processing_queues)
  conn.hdel(REGISTRY_KEY, *processing_queues) unless processing_queues.empty?
end