Skip to content

Commit d080906

Browse files
committed
Prevent ThreadedWorker from leaking enqueue lifecycle callbacks
Delayed::Plugin callbacks register against Delayed::Worker.lifecycle (the base class), and Delayed::Job.enqueue runs that same base lifecycle. Delayed::Worker#initialize calls self.class.setup_lifecycle, which rebuilds the lifecycle on the class it is called on. Since ThreadedWorker is a subclass, setup_lifecycle reset a different lifecycle than the base one the plugins append to. The base lifecycle was therefore never reset, and every ThreadedWorker.new added another before(:enqueue) callback to it. The before(:enqueue) callback creates a PollableJobModel row, so the accumulation caused a single enqueue to create many pollable rows. Production is unaffected since a worker is created once per process. Delegate setup_lifecycle and lifecycle on ThreadedWorker to the base class so each instantiation rebuilds the same lifecycle cleanly.
1 parent 12e5d15 commit d080906

2 files changed

Lines changed: 23 additions & 0 deletions

File tree

lib/delayed_job/threaded_worker.rb

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,18 @@
11
module Delayed
22
class ThreadedWorker < Delayed::Worker
3+
# Delayed::Plugin callbacks register against the base Delayed::Worker.lifecycle,
4+
# but setup_lifecycle rebuilds the lifecycle on the class it is called on. On this
5+
# subclass it would reset a different lifecycle than the one plugins append to,
6+
# so every ThreadedWorker.new would leak another callback into the base lifecycle.
7+
# Delegate to the base class so each instantiation rebuilds the same lifecycle.
8+
def self.setup_lifecycle
9+
Delayed::Worker.setup_lifecycle
10+
end
11+
12+
def self.lifecycle
13+
Delayed::Worker.lifecycle
14+
end
15+
316
def initialize(options={})
417
super
518
@num_threads = options[:num_threads]

spec/unit/lib/delayed_job/threaded_worker_spec.rb

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,16 @@
2222
worker = Delayed::ThreadedWorker.new({ num_threads: 2 })
2323
expect(worker.instance_variable_get(:@grace_period_seconds)).to eq(30)
2424
end
25+
26+
it 'does not accumulate enqueue lifecycle callbacks across instantiations' do
27+
before_enqueue_callbacks = lambda do
28+
Delayed::Worker.lifecycle.instance_variable_get(:@callbacks)[:enqueue].instance_variable_get(:@before).size
29+
end
30+
31+
baseline = before_enqueue_callbacks.call
32+
5.times { Delayed::ThreadedWorker.new(options) }
33+
expect(before_enqueue_callbacks.call).to eq(baseline)
34+
end
2535
end
2636

2737
describe '#start' do

0 commit comments

Comments
 (0)