Module: RubyReactor::SweeperJob

Included in:
Adapters::ActiveJob::SweeperWorker, Adapters::Sidekiq::SweeperWorker
Defined in:
lib/ruby_reactor/sweeper_job.rb

Overview

Self-rescheduling recovery tick, shared by every queueing backend's sweeper job class. Each run sweeps both the top-level reactor sweeper and the map sweeper, then schedules the next tick — a perpetual chain the host kicks once via RubyReactor.start_sweeper!.

super_fetch safety. Sidekiq Enterprise super_fetch reliably re-runs a job whose worker died mid-execution. For a self-rescheduling chain that is a hazard: a tick can crash AFTER enqueuing its successor but BEFORE acking, so super_fetch recovers the crashed tick alongside the successor it already scheduled — the chain forks and then doubles every interval. We therefore do NOT rely on "exactly one job exists". The next tick is claimed by a per-time-window lock: every duplicate computes the SAME target window and only one wins the claim, so recovered/duplicated ticks collapse back to a single chain. The claim lock is never released — it simply expires — so no delete can race two duplicates into both winning.

Defined Under Namespace

Modules: ClassMethods

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.included(base) ⇒ Object 



22
23
24
 
# File 'lib/ruby_reactor/sweeper_job.rb', line 22

def self.included(base)
  base.extend(ClassMethods)
end

Instance Method Details

#perform ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/ruby_reactor/sweeper_job.rb', line 26

def perform
  config = RubyReactor.configuration
  return unless config.sweeper_enabled

  run_sweeps(config)
ensure
  # Always chain forward (unless disabled), even after an error above, so a
  # single bad sweep can't kill recovery. The window lock keeps this from
  # forking under super_fetch.
  self.class.schedule_next if RubyReactor.configuration.sweeper_enabled
end

#run_sweeps(config) ⇒ Object 



38
39
40
41
42
43
 
# File 'lib/ruby_reactor/sweeper_job.rb', line 38

def run_sweeps(config)
  RubyReactor::Sweeper.run_once(limit: config.sweeper_limit)
  RubyReactor::Map::Sweeper.run_once(limit: config.sweeper_limit)
rescue StandardError => e
  config.logger.error("RubyReactor sweeper sweep failed: #{e.class}: #{e.message}")
end