Permalink
Browse files

Retry rate limited jobs more aggressively

  • Loading branch information...
1 parent eb5c064 commit 07df5d75a6053b723287ebf67c0054978e0ad944 @gylaz gylaz committed Apr 26, 2013
@@ -1,31 +1,17 @@
-# Delayed job for posting activity stories to the Yammer API endpoint
-
class ActivityCreatorJob < Struct.new(:user_id, :action, :event_id)
include Rails.application.routes.url_helpers
-
- PRIORITY = 1
+ include YammerRateLimited
def self.enqueue(user, action, event)
- Delayed::Job.enqueue(
- new(user.id, action, event.id),
- priority: PRIORITY
- )
+ job = new(user.id, action, event.id)
+
+ Delayed::Job.enqueue(job)
end
def perform
post_yammer_activity
end
- def error(job, exception)
- unless ExceptionSilencer.is_rate_limit?(exception)
- Airbrake.notify(exception)
- end
- end
-
- def failure(job)
- Airbrake.notify(error_message: "Job failure: #{job.last_error}")
- end
-
private
def user
@@ -1,9 +1,8 @@
class EventCreatedEmailJob < Struct.new(:event_id)
- PRIORITY = 1
- ACTION = 'create'
-
def self.enqueue(event)
- Delayed::Job.enqueue(new(event.id), priority: PRIORITY)
+ job = new(event.id)
+
+ Delayed::Job.enqueue(job)
end
def error(job, exception)
@@ -1,24 +1,16 @@
class InvitationCreatedMessageJob < Struct.new(:invitation_id)
- PRIORITY = 1
+ include YammerRateLimited
def self.enqueue(invitation)
- Delayed::Job.enqueue(new(invitation.id), priority: PRIORITY)
+ job = new(invitation.id)
+
+ Delayed::Job.enqueue(job)
end
def perform
invitee.invite(invitation)
end
- def error(job, exception)
- unless ExceptionSilencer.is_rate_limit?(exception)
- Airbrake.notify(exception)
- end
- end
-
- def failure(job)
- Airbrake.notify(error_message: "Job failure: #{job.last_error}")
- end
-
private
def invitation
@@ -1,27 +1,14 @@
class ReminderCreatedJob < Struct.new(:reminder_id)
- PRIORITY = 1
+ include YammerRateLimited
def self.enqueue(reminder)
- Delayed::Job.enqueue(new(reminder.id), priority: PRIORITY)
+ job = new(reminder.id)
+
+ Delayed::Job.enqueue(job)
end
def perform
+ reminder = Reminder.find(reminder_id)
reminder.deliver
end
-
- def error(job, exception)
- unless ExceptionSilencer.is_rate_limit?(exception)
- Airbrake.notify(exception)
- end
- end
-
- def failure(job)
- Airbrake.notify(error_message: "Job failure: #{job.last_error}")
- end
-
- private
-
- def reminder
- @reminder ||= Reminder.find(reminder_id)
- end
end
@@ -1,9 +1,6 @@
class VoteCreatedJob < Struct.new(:vote_id)
- PRIORITY = 1
- ACTION = 'vote'
-
def self.enqueue(vote)
- Delayed::Job.enqueue(new(vote.id), priority: PRIORITY)
+ Delayed::Job.enqueue(new(vote.id))
end
def perform
@@ -26,7 +23,7 @@ def vote
def create_activity_message
if voter.yammer_user?
- ActivityCreatorJob.enqueue(voter, ACTION, event)
+ ActivityCreatorJob.enqueue(voter, 'vote', event)
end
end
View
@@ -1,13 +1,10 @@
class VoteEmailJob < Struct.new(:vote_id, :email_type)
- PRIORITY = 1
DELAY = 3.minutes
def self.enqueue(vote, email_type)
- Delayed::Job.enqueue(
- new(vote.id, email_type),
- run_at: DELAY.from_now,
- priority: PRIORITY
- )
+ job = new(vote.id, email_type)
+
+ Delayed::Job.enqueue(job, run_at: DELAY.from_now)
end
def perform
@@ -32,12 +29,11 @@ def find_vote
def log_error_and_return_false
log_error
- return false
+ false
end
def log_error
- Rails.logger.
- error "NOTE: VoteEmailJob cannot find Vote id=#{vote_id}"
+ Rails.logger.error("NOTE: VoteEmailJob cannot find Vote id=#{vote_id}")
end
def no_other_recent_votes
@@ -0,0 +1,26 @@
+module YammerRateLimited
+ RATE_LIMIT_RETRIES = 50
+
+ def error(job, exception)
+ if ExceptionSilencer.is_rate_limit?(exception)
+ @rate_limited = true
+ else
+ @rate_limited = false
+ Airbrake.notify(exception)
+ end
+ end
+
+ def failure(job)
+ Airbrake.notify(error_message: "Job failure: #{job.last_error}")
+ end
+
+ def reschedule_at(attempts, time)
+ if @rate_limited
+ 30.seconds.from_now
+ end
+ end
+
+ def max_attempts
+ @rate_limited ? RATE_LIMIT_RETRIES : Delayed::Worker.max_attempts
+ end
+end
@@ -7,12 +7,10 @@
event = build_stubbed(:event)
Delayed::Job.stubs(:enqueue)
activity_creator_job = ActivityCreatorJob.new(user.id, action, event.id)
- priority = 1
ActivityCreatorJob.enqueue(user, action, event)
- expect(Delayed::Job).to have_received(:enqueue).
- with(activity_creator_job, priority: priority)
+ expect(Delayed::Job).to have_received(:enqueue).with(activity_creator_job)
end
end
@@ -5,12 +5,10 @@
event = build_stubbed(:event)
Delayed::Job.stubs(:enqueue)
event_created_email_job = EventCreatedEmailJob.new(event.id)
- priority = 1
EventCreatedEmailJob.enqueue(event)
- expect(Delayed::Job).to have_received(:enqueue).
- with(event_created_email_job, priority: priority)
+ expect(Delayed::Job).to have_received(:enqueue).with(event_created_email_job)
end
end
@@ -5,14 +5,11 @@
invitation = build_stubbed(:invitation)
Invitation.stubs(find: invitation)
Delayed::Job.stubs(:enqueue)
- invitation_created_message_job =
- InvitationCreatedMessageJob.new(invitation.id)
- priority = 1
+ job = InvitationCreatedMessageJob.new(invitation.id)
InvitationCreatedMessageJob.enqueue(invitation)
- expect(Delayed::Job).to have_received(:enqueue).
- with(invitation_created_message_job, priority: priority)
+ expect(Delayed::Job).to have_received(:enqueue).with(job)
end
end
@@ -39,16 +36,6 @@
expect(Airbrake).to have_received(:notify).with(exception)
end
-
- it 'does not send exception to Airbrake if the job errors due to rate limit' do
- job = ReminderCreatedJob.new
- exception = Faraday::Error::ClientError.new('Rate limited!', status: 429)
- Airbrake.stubs(:notify)
-
- job.error(job, exception)
-
- expect(Airbrake).to have_received(:notify).never
- end
end
describe InvitationCreatedMessageJob, '#failure' do
@@ -63,3 +50,36 @@
with(error_message: 'Job failure: boom')
end
end
+
+describe InvitationCreatedMessageJob, '#max_attempts' do
+ it 'uses the global config settings' do
+ job = InvitationCreatedMessageJob.new
+
+ expect(job.max_attempts).to eq 5
+ end
+
+ context 'when rate limit error is encountered' do
+ it 'overrides global delayed_job settings' do
+ exception = Faraday::Error::ClientError.new('Rate limited!', status: 429)
+ job = InvitationCreatedMessageJob.new
+ job.error(job, exception)
+
+ expect(job.max_attempts).to eq 50
+ end
+ end
+end
+
+describe InvitationCreatedMessageJob, '#reschedule_at' do
+ context 'when rate limit error is encountered' do
+ it 'overrides global delayed_job settings' do
+ ignored = nil
+ exception = Faraday::Error::ClientError.new('Rate limited!', status: 429)
+ job = InvitationCreatedMessageJob.new
+ job.error(job, exception)
+
+ result = job.reschedule_at(ignored, ignored)
+
+ expect(result).to be_within(1).of(30.seconds.from_now)
+ end
+ end
+end
@@ -6,12 +6,10 @@
Reminder.stubs(find: reminder)
Delayed::Job.stubs(:enqueue)
reminder_created_job = ReminderCreatedJob.new(reminder.id)
- priority = 1
ReminderCreatedJob.enqueue(reminder)
- expect(Delayed::Job).to have_received(:enqueue).
- with(reminder_created_job, priority: priority)
+ expect(Delayed::Job).to have_received(:enqueue).with(reminder_created_job)
end
end
@@ -6,12 +6,10 @@
Vote.stubs(find: vote)
Delayed::Job.stubs(:enqueue)
vote_created_job = VoteCreatedJob.new(vote.id)
- priority = 1
VoteCreatedJob.enqueue(vote)
- expect(Delayed::Job).to have_received(:enqueue).
- with(vote_created_job, priority: priority)
+ expect(Delayed::Job).to have_received(:enqueue).with(vote_created_job)
end
end
@@ -6,13 +6,12 @@
vote = build_stubbed(:vote)
Delayed::Job.stubs(:enqueue)
vote_email_job = VoteEmailJob.new(vote.id, :test_vote_email)
- priority = 1
delay = 3.minutes
VoteEmailJob.enqueue(vote, :test_vote_email)
expect(Delayed::Job).to have_received(:enqueue).
- with(vote_email_job, priority: priority, run_at: delay.from_now)
+ with(vote_email_job, run_at: delay.from_now)
end
end
end
View
@@ -52,10 +52,11 @@
voter = first_vote.voter
event = first_vote.suggestion.event
second_suggestion = create(:suggestion, event: event)
-
- Timecop.freeze(VoteEmailJob::DELAY.from_now)
- second_vote = create(:vote, voter: voter, suggestion: second_suggestion)
- Timecop.return
+ second_vote = create(:vote,
+ voter: voter,
+ suggestion: second_suggestion,
+ created_at: VoteEmailJob::DELAY.from_now
+ )
first_vote_check = first_vote.has_no_other_votes_within_delay_window?
second_vote_check = second_vote.has_no_other_votes_within_delay_window?

0 comments on commit 07df5d7

Please sign in to comment.