collectiveidea / delayed_job forked from tobi/delayed_job

Database based asynchronously priority queue system -- Extracted from Shopify

delayed_job / spec / job_spec.rb
100644 346 lines (260 sloc) 11.791 kb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
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
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
require File.dirname(__FILE__) + '/database'
 
class SimpleJob
  cattr_accessor :runs; self.runs = 0
  def perform; @@runs += 1; end
end
 
class ErrorJob
  cattr_accessor :runs; self.runs = 0
  def perform; raise 'did not work'; end
end
 
module M
  class ModuleJob
    cattr_accessor :runs; self.runs = 0
    def perform; @@runs += 1; end
  end
  
end
 
describe Delayed::Job do
  before do
    Delayed::Job.max_priority = nil
    Delayed::Job.min_priority = nil
    
    Delayed::Job.delete_all
  end
  
  before(:each) do
    SimpleJob.runs = 0
  end
 
  it "should set run_at automatically if not set" do
    Delayed::Job.create(:payload_object => ErrorJob.new ).run_at.should_not == nil
  end
 
  it "should not set run_at automatically if already set" do
    later = 5.minutes.from_now
    Delayed::Job.create(:payload_object => ErrorJob.new, :run_at => later).run_at.should == later
  end
 
  it "should raise ArgumentError when handler doesn't respond_to :perform" do
    lambda { Delayed::Job.enqueue(Object.new) }.should raise_error(ArgumentError)
  end
 
  it "should increase count after enqueuing items" do
    Delayed::Job.enqueue SimpleJob.new
    Delayed::Job.count.should == 1
  end
 
  it "should be able to set priority when enqueuing items" do
    Delayed::Job.enqueue SimpleJob.new, 5
    Delayed::Job.first.priority.should == 5
  end
 
  it "should be able to set run_at when enqueuing items" do
    later = 5.minutes.from_now
    Delayed::Job.enqueue SimpleJob.new, 5, later
 
    # use be close rather than equal to because millisecond values cn be lost in DB round trip
    Delayed::Job.first.run_at.should be_close(later, 1)
  end
 
  it "should call perform on jobs when running work_off" do
    SimpleJob.runs.should == 0
 
    Delayed::Job.enqueue SimpleJob.new
    Delayed::Job.work_off
 
    SimpleJob.runs.should == 1
  end
                     
                     
  it "should work with eval jobs" do
    $eval_job_ran = false
 
    Delayed::Job.enqueue do <<-JOB
$eval_job_ran = true
JOB
    end
 
    Delayed::Job.work_off
 
    $eval_job_ran.should == true
  end
                   
  it "should work with jobs in modules" do
    M::ModuleJob.runs.should == 0
 
    Delayed::Job.enqueue M::ModuleJob.new
    Delayed::Job.work_off
 
    M::ModuleJob.runs.should == 1
  end
                   
  it "should re-schedule by about 1 second at first and increment this more and more minutes when it fails to execute properly" do
    Delayed::Job.enqueue ErrorJob.new
    Delayed::Job.work_off(1)
 
    job = Delayed::Job.find(:first)
 
    job.last_error.should =~ /did not work/
    job.last_error.should =~ /job_spec.rb:10:in `perform'/
    job.attempts.should == 1
 
    job.run_at.should > Delayed::Job.db_time_now - 10.minutes
    job.run_at.should < Delayed::Job.db_time_now + 10.minutes
  end
 
  it "should raise an DeserializationError when the job class is totally unknown" do
 
    job = Delayed::Job.new
    job['handler'] = "--- !ruby/object:JobThatDoesNotExist {}"
 
    lambda { job.payload_object.perform }.should raise_error(Delayed::DeserializationError)
  end
 
  it "should try to load the class when it is unknown at the time of the deserialization" do
    job = Delayed::Job.new
    job['handler'] = "--- !ruby/object:JobThatDoesNotExist {}"
 
    job.should_receive(:attempt_to_load).with('JobThatDoesNotExist').and_return(true)
 
    lambda { job.payload_object.perform }.should raise_error(Delayed::DeserializationError)
  end
 
  it "should try include the namespace when loading unknown objects" do
    job = Delayed::Job.new
    job['handler'] = "--- !ruby/object:Delayed::JobThatDoesNotExist {}"
    job.should_receive(:attempt_to_load).with('Delayed::JobThatDoesNotExist').and_return(true)
    lambda { job.payload_object.perform }.should raise_error(Delayed::DeserializationError)
  end
 
  it "should also try to load structs when they are unknown (raises TypeError)" do
    job = Delayed::Job.new
    job['handler'] = "--- !ruby/struct:JobThatDoesNotExist {}"
 
    job.should_receive(:attempt_to_load).with('JobThatDoesNotExist').and_return(true)
 
    lambda { job.payload_object.perform }.should raise_error(Delayed::DeserializationError)
  end
 
  it "should try include the namespace when loading unknown structs" do
    job = Delayed::Job.new
    job['handler'] = "--- !ruby/struct:Delayed::JobThatDoesNotExist {}"
 
    job.should_receive(:attempt_to_load).with('Delayed::JobThatDoesNotExist').and_return(true)
    lambda { job.payload_object.perform }.should raise_error(Delayed::DeserializationError)
  end
  
  it "should be failed if it failed more than MAX_ATTEMPTS times and we don't want to destroy jobs" do
    default = Delayed::Job.destroy_failed_jobs
    Delayed::Job.destroy_failed_jobs = false
 
    @job = Delayed::Job.create :payload_object => SimpleJob.new, :attempts => 50
    @job.reload.failed_at.should == nil
    @job.reschedule 'FAIL'
    @job.reload.failed_at.should_not == nil
 
    Delayed::Job.destroy_failed_jobs = default
  end
 
  it "should be destroyed if it failed more than MAX_ATTEMPTS times and we want to destroy jobs" do
    default = Delayed::Job.destroy_failed_jobs
    Delayed::Job.destroy_failed_jobs = true
 
    @job = Delayed::Job.create :payload_object => SimpleJob.new, :attempts => 50
    @job.should_receive(:destroy)
    @job.reschedule 'FAIL'
 
    Delayed::Job.destroy_failed_jobs = default
  end
 
  it "should never find failed jobs" do
    @job = Delayed::Job.create :payload_object => SimpleJob.new, :attempts => 50, :failed_at => Time.now
    Delayed::Job.find_available(1).length.should == 0
  end
 
  context "when another worker is already performing an task, it" do
 
    before :each do
      Delayed::Job.worker_name = 'worker1'
      @job = Delayed::Job.create :payload_object => SimpleJob.new, :locked_by => 'worker1', :locked_at => Delayed::Job.db_time_now - 5.minutes
    end
 
    it "should not allow a second worker to get exclusive access" do
      @job.lock_exclusively!(4.hours, 'worker2').should == false
    end
 
    it "should allow a second worker to get exclusive access if the timeout has passed" do
      @job.lock_exclusively!(1.minute, 'worker2').should == true
    end
    
    it "should be able to get access to the task if it was started more then max_age ago" do
      @job.locked_at = 5.hours.ago
      @job.save
 
      @job.lock_exclusively! 4.hours, 'worker2'
      @job.reload
      @job.locked_by.should == 'worker2'
      @job.locked_at.should > 1.minute.ago
    end
 
    it "should not be found by another worker" do
      Delayed::Job.worker_name = 'worker2'
 
      Delayed::Job.find_available(1, 6.minutes).length.should == 0
    end
 
    it "should be found by another worker if the time has expired" do
      Delayed::Job.worker_name = 'worker2'
 
      Delayed::Job.find_available(1, 4.minutes).length.should == 1
    end
 
    it "should be able to get exclusive access again when the worker name is the same" do
      @job.lock_exclusively! 5.minutes, 'worker1'
      @job.lock_exclusively! 5.minutes, 'worker1'
      @job.lock_exclusively! 5.minutes, 'worker1'
    end
  end
  
  context "#name" do
    it "should be the class name of the job that was enqueued" do
      Delayed::Job.create(:payload_object => ErrorJob.new ).name.should == 'ErrorJob'
    end
 
    it "should be the method that will be called if its a performable method object" do
      Delayed::Job.send_later(:clear_locks!)
      Delayed::Job.last.name.should == 'Delayed::Job.clear_locks!'
 
    end
    it "should be the instance method that will be called if its a performable method object" do
      story = Story.create :text => "..."
      
      story.send_later(:save)
      
      Delayed::Job.last.name.should == 'Story#save'
    end
  end
  
  context "worker prioritization" do
    
    before(:each) do
      Delayed::Job.max_priority = nil
      Delayed::Job.min_priority = nil
    end
  
    it "should only work_off jobs that are >= min_priority" do
      Delayed::Job.min_priority = -5
      Delayed::Job.max_priority = 5
      SimpleJob.runs.should == 0
    
      Delayed::Job.enqueue SimpleJob.new, -10
      Delayed::Job.enqueue SimpleJob.new, 0
      Delayed::Job.work_off
    
      SimpleJob.runs.should == 1
    end
  
    it "should only work_off jobs that are <= max_priority" do
      Delayed::Job.min_priority = -5
      Delayed::Job.max_priority = 5
      SimpleJob.runs.should == 0
    
      Delayed::Job.enqueue SimpleJob.new, 10
      Delayed::Job.enqueue SimpleJob.new, 0
 
      Delayed::Job.work_off
 
      SimpleJob.runs.should == 1
    end
   
  end
  
  context "when pulling jobs off the queue for processing, it" do
    before(:each) do
      @job = Delayed::Job.create(
        :payload_object => SimpleJob.new,
        :locked_by => 'worker1',
        :locked_at => Delayed::Job.db_time_now - 5.minutes)
    end
 
    it "should leave the queue in a consistent state and not run the job if locking fails" do
      SimpleJob.runs.should == 0
      @job.stub!(:lock_exclusively!).with(any_args).once.and_return(false)
      Delayed::Job.should_receive(:find_available).once.and_return([@job])
      Delayed::Job.work_off(1)
      SimpleJob.runs.should == 0
    end
  
  end
  
  context "while running alongside other workers that locked jobs, it" do
    before(:each) do
      Delayed::Job.worker_name = 'worker1'
      Delayed::Job.create(:payload_object => SimpleJob.new, :locked_by => 'worker1', :locked_at => (Delayed::Job.db_time_now - 1.minutes))
      Delayed::Job.create(:payload_object => SimpleJob.new, :locked_by => 'worker2', :locked_at => (Delayed::Job.db_time_now - 1.minutes))
      Delayed::Job.create(:payload_object => SimpleJob.new)
      Delayed::Job.create(:payload_object => SimpleJob.new, :locked_by => 'worker1', :locked_at => (Delayed::Job.db_time_now - 1.minutes))
    end
 
    it "should ingore locked jobs from other workers" do
      Delayed::Job.worker_name = 'worker3'
      SimpleJob.runs.should == 0
      Delayed::Job.work_off
      SimpleJob.runs.should == 1 # runs the one open job
    end
 
    it "should find our own jobs regardless of locks" do
      Delayed::Job.worker_name = 'worker1'
      SimpleJob.runs.should == 0
      Delayed::Job.work_off
      SimpleJob.runs.should == 3 # runs open job plus worker1 jobs that were already locked
    end
  end
 
  context "while running with locked and expired jobs, it" do
    before(:each) do
      Delayed::Job.worker_name = 'worker1'
      exp_time = Delayed::Job.db_time_now - (1.minutes + Delayed::Job::MAX_RUN_TIME)
      Delayed::Job.create(:payload_object => SimpleJob.new, :locked_by => 'worker1', :locked_at => exp_time)
      Delayed::Job.create(:payload_object => SimpleJob.new, :locked_by => 'worker2', :locked_at => (Delayed::Job.db_time_now - 1.minutes))
      Delayed::Job.create(:payload_object => SimpleJob.new)
      Delayed::Job.create(:payload_object => SimpleJob.new, :locked_by => 'worker1', :locked_at => (Delayed::Job.db_time_now - 1.minutes))
    end
 
    it "should only find unlocked and expired jobs" do
      Delayed::Job.worker_name = 'worker3'
      SimpleJob.runs.should == 0
      Delayed::Job.work_off
      SimpleJob.runs.should == 2 # runs the one open job and one expired job
    end
 
    it "should ignore locks when finding our own jobs" do
      Delayed::Job.worker_name = 'worker1'
      SimpleJob.runs.should == 0
      Delayed::Job.work_off
      SimpleJob.runs.should == 3 # runs open job plus worker1 jobs
      # This is useful in the case of a crash/restart on worker1, but make sure multiple workers on the same host have unique names!
    end
 
  end
  
end