ms precision for settings and exponential back off
transformed time based values in settings to float to gain ms
resolution
added setting for fuzzied exponential back off
MODIFY
.gitignore
.gitignore
@@ -24,3 +24,5 @@ TAGS
/go/src/gopkg.in/tylerb/graceful.v1/
/go/src/gopkg.in/yaml.v2/
/go/src/source.xing.com/olympus/golympus/
+.idea
+.byebug_history

MODIFY
lib/beetle/message.rb
lib/beetle/message.rb
@@ -13,14 +13,14 @@ module Beetle
# flag for encoding redundant messages
FLAG_REDUNDANT = 1
# default lifetime of messages
- DEFAULT_TTL = 1.day
+ DEFAULT_TTL = 86_400
# forcefully abort a running handler after this many seconds.
# can be overriden when registering a handler.
- DEFAULT_HANDLER_TIMEOUT = 600.seconds
+ DEFAULT_HANDLER_TIMEOUT = 600.0
# how many times we should try to run a handler before giving up
DEFAULT_HANDLER_EXECUTION_ATTEMPTS = 1
# how many seconds we should wait before retrying handler execution
- DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY = 10.seconds
+ DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY = 10.0
# how many exceptions should be tolerated before giving up
DEFAULT_EXCEPTION_LIMIT = 0
@@ -67,6 +67,7 @@ module Beetle
@server = opts[:server]
@timeout = opts[:timeout] || DEFAULT_HANDLER_TIMEOUT
@delay = opts[:delay] || DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY
+ @exp_bo = !!opts[:exponential_back_off]
@attempts_limit = opts[:attempts] || DEFAULT_HANDLER_EXECUTION_ATTEMPTS
@exceptions_limit = opts[:exceptions] || DEFAULT_EXCEPTION_LIMIT
@attempts_limit = @exceptions_limit + 1 if @attempts_limit <= @exceptions_limit
@@ -126,12 +127,12 @@ module Beetle
# current time (UNIX timestamp)
def now #:nodoc:
- Time.now.to_i
+ Time.now.to_f
end
# current time (UNIX timestamp)
def self.now #:nodoc:
- Time.now.to_i
+ Time.now.to_f
end
# a message has expired if the header expiration timestamp is msaller than the current time
@@ -161,7 +162,7 @@ module Beetle
# handler timed out?
def timed_out?
- (t = @store.get(msg_id, :timeout)) && t.to_i < now
+ (t = @store.get(msg_id, :timeout)) && t.to_f < now
end
# reset handler timeout in the deduplication store
@@ -181,12 +182,13 @@ module Beetle
# whether we should wait before running the handler
def delayed?
- (t = @store.get(msg_id, :delay)) && t.to_i > now
+ (t = @store.get(msg_id, :delay)) && t.to_f > now
end
# store delay value in the deduplication store
def set_delay!
- @store.set(msg_id, :delay, now + delay)
+ next_delay = calculate_next_delay(attempts, delay)
+ @store.set(msg_id, :delay, now + next_delay)
end
# how many times we already tried running the handler
@@ -281,10 +283,10 @@ module Beetle
if status == "completed"
ack!
RC::OK
- elsif delay && delay.to_i > now
+ elsif delay && delay.to_f > now
logger.warn "Beetle: ignored delayed message (#{msg_id})!"
RC::Delayed
- elsif !(timeout && timeout.to_i < now)
+ elsif !(timeout && timeout.to_f < now)
RC::HandlerNotYetTimedOut
elsif attempts.to_i >= attempts_limit
completed!
@@ -362,5 +364,18 @@ module Beetle
@store.del_keys(msg_id)
end
end
+
+ def exponential_back_off?
+ @exp_bo
+ end
+
+ def calculate_next_delay(n, base_delay)
+ unless exponential_back_off?
+ base_delay
+ else
+ base_delay_n = base_delay * (2**n)
+ rand((base_delay_n - base_delay_n / 4.0)..(base_delay_n + base_delay_n / 2.0))
+ end
+ end
end
end

ADD
test/beetle/message/settings_test.rb
+require File.expand_path(File.dirname(__FILE__) + '/../../test_helper')
+
+module Beetle
+ class SettingsTest < Minitest::Test
+ def setup
+ @store = DeduplicationStore.new
+ @store.flushdb
+ end
+
+ test "completed! should store the status 'complete' in the database" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ assert !message.completed?
+ message.completed!
+ assert message.completed?
+ assert_equal "completed", @store.get(message.msg_id, :status)
+ end
+
+ test "set_delay! should store the current time plus the delay offset in the database" do
+ message = Message.new("somequeue", header_with_params, 'foo', :delay => 1.75, :store => @store)
+ message.expects(:now).returns(1.2)
+ message.set_delay!
+ assert_equal "2.95", @store.get(message.msg_id, :delay)
+ message.expects(:now).returns(3.0)
+ assert !message.delayed?
+ message.expects(:now).returns(0.0)
+ assert message.delayed?
+ end
+
+ test "set_delay! should store the current time plus the exponential delay offset in the database" do
+ message = Message.new("somequeue", header_with_params, 'foo', :delay => 1.0, :exponential_back_off => true, :store => @store)
+ [0.0, 1.5, 3.33].each do |now_offset|
+ message.stubs(:now).returns(now_offset)
+ [
+ [0.75, 1.5],
+ [1.5, 3.0],
+ [3.0, 6.0],
+ [6.0, 12.0],
+ ].each do |(lower, upper)|
+ message.set_delay!
+ assert @store.get(message.msg_id, :delay).to_f.between?(lower + now_offset, upper + now_offset)
+ message.increment_execution_attempts!
+ end
+ @store.set(message.msg_id, :attempts, 0)
+ end
+ end
+
+ test "set_delay! should store the current time plus the linear delay offset in the database" do
+ message = Message.new("somequeue", header_with_params, 'foo', :delay => 0.4, :exponential_back_off => false, :store => @store)
+ [
+ [1.6, 2.0],
+ [2.1, 2.5],
+ [2.7, 3.1],
+ ].each do |now_offset, exp_delay|
+ message.stubs(:now).returns(now_offset)
+ message.set_delay!
+ assert_equal @store.get(message.msg_id, :delay).to_f, exp_delay
+ message.increment_execution_attempts!
+ end
+ end
+
+ test "set_delay! should use the default delay if the delay hasn't been set on the message instance" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ message.expects(:now).returns(0.0)
+ message.set_delay!
+ assert_equal "#{Message::DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY}", @store.get(message.msg_id, :delay)
+ message.expects(:now).returns(message.delay)
+ assert !message.delayed?
+ message.expects(:now).returns(0.0)
+ assert message.delayed?
+ end
+
+ test "set_timeout! should store the current time plus the number of timeout seconds in the database" do
+ message = Message.new("somequeue", header_with_params, 'foo', :timeout => 1, :store => @store)
+ message.expects(:now).returns(1.0)
+ message.set_timeout!
+ assert_equal "2.0", @store.get(message.msg_id, :timeout)
+ message.expects(:now).returns(2.0)
+ assert !message.timed_out?
+ message.expects(:now).returns(3.0)
+ assert message.timed_out?
+ end
+
+ test "set_timeout! should use the default timeout if the timeout hasn't been set on the message instance" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ message.expects(:now).returns(0.0)
+ message.set_timeout!
+ assert_equal "#{Message::DEFAULT_HANDLER_TIMEOUT}", @store.get(message.msg_id, :timeout)
+ message.expects(:now).returns(message.timeout)
+ assert !message.timed_out?
+ message.expects(:now).returns(Message::DEFAULT_HANDLER_TIMEOUT + 1.0)
+ assert message.timed_out?
+ end
+
+ test "incrementing execution attempts should increment by 1" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ assert_equal 1, message.increment_execution_attempts!
+ assert_equal 2, message.increment_execution_attempts!
+ assert_equal 3, message.increment_execution_attempts!
+ end
+
+ test "accessing execution attempts should return the number of execution attempts made so far" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ assert_equal 0, message.attempts
+ message.increment_execution_attempts!
+ assert_equal 1, message.attempts
+ message.increment_execution_attempts!
+ assert_equal 2, message.attempts
+ message.increment_execution_attempts!
+ assert_equal 3, message.attempts
+ end
+
+ test "accessing execution attempts should return 0 if none were made" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ assert_equal 0, message.attempts
+ end
+
+
+ test "attempts limit should be set exception limit + 1 iff the configured attempts limit is equal to or smaller than the exceptions limit" do
+ message = Message.new("somequeue", header_with_params, 'foo', :exceptions => 1, :store => @store)
+ assert_equal 2, message.attempts_limit
+ assert_equal 1, message.exceptions_limit
+ message = Message.new("somequeue", header_with_params, 'foo', :exceptions => 2, :store => @store)
+ assert_equal 3, message.attempts_limit
+ assert_equal 2, message.exceptions_limit
+ message = Message.new("somequeue", header_with_params, 'foo', :attempts => 5, :exceptions => 2, :store => @store)
+ assert_equal 5, message.attempts_limit
+ assert_equal 2, message.exceptions_limit
+ end
+
+ test "attempts limit should be reached after incrementing the attempt limit counter 'attempts limit' times" do
+ message = Message.new("somequeue", header_with_params, 'foo', :attempts =>2, :store => @store)
+ assert !message.attempts_limit_reached?
+ message.increment_execution_attempts!
+ assert !message.attempts_limit_reached?
+ message.increment_execution_attempts!
+ assert message.attempts_limit_reached?
+ message.increment_execution_attempts!
+ assert message.attempts_limit_reached?
+ end
+
+ test "incrementing exception counts should increment by 1" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ assert_equal 1, message.increment_exception_count!
+ assert_equal 2, message.increment_exception_count!
+ assert_equal 3, message.increment_exception_count!
+ end
+
+ test "default exceptions limit should be reached after incrementing the attempt limit counter 1 time" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ assert !message.exceptions_limit_reached?
+ message.increment_exception_count!
+ assert message.exceptions_limit_reached?
+ end
+
+ test "exceptions limit should be reached after incrementing the attempt limit counter 'exceptions limit + 1' times" do
+ message = Message.new("somequeue", header_with_params, 'foo', :exceptions => 1, :store => @store)
+ assert !message.exceptions_limit_reached?
+ message.increment_exception_count!
+ assert !message.exceptions_limit_reached?
+ message.increment_exception_count!
+ assert message.exceptions_limit_reached?
+ message.increment_exception_count!
+ assert message.exceptions_limit_reached?
+ end
+
+ test "failure to aquire a mutex should delete it from the database" do
+ message = Message.new("somequeue", header_with_params, 'foo', :store => @store)
+ assert message.aquire_mutex!
+ assert !message.aquire_mutex!
+ assert !@store.exists(message.msg_id, :mutex)
+ end
+ end
+end

MODIFY
test/beetle/message_test.rb
test/beetle/message_test.rb
@@ -677,159 +677,6 @@ module Beetle
end
end
- class SettingsTest < Minitest::Test
- def setup
- @store = DeduplicationStore.new
- @store.flushdb
- end
-
- test "completed! should store the status 'complete' in the database" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- assert !message.completed?
- message.completed!
- assert message.completed?
- assert_equal "completed", @store.get(message.msg_id, :status)
- end
-
- test "set_delay! should store the current time plus the number of delayed seconds in the database" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :delay => 1, :store => @store)
- message.expects(:now).returns(1)
- message.set_delay!
- assert_equal "2", @store.get(message.msg_id, :delay)
- message.expects(:now).returns(2)
- assert !message.delayed?
- message.expects(:now).returns(0)
- assert message.delayed?
- end
-
- test "set_delay! should use the default delay if the delay hasn't been set on the message instance" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- message.expects(:now).returns(0)
- message.set_delay!
- assert_equal "#{Message::DEFAULT_HANDLER_EXECUTION_ATTEMPTS_DELAY}", @store.get(message.msg_id, :delay)
- message.expects(:now).returns(message.delay)
- assert !message.delayed?
- message.expects(:now).returns(0)
- assert message.delayed?
- end
-
- test "set_timeout! should store the current time plus the number of timeout seconds in the database" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :timeout => 1, :store => @store)
- message.expects(:now).returns(1)
- message.set_timeout!
- assert_equal "2", @store.get(message.msg_id, :timeout)
- message.expects(:now).returns(2)
- assert !message.timed_out?
- message.expects(:now).returns(3)
- assert message.timed_out?
- end
-
- test "set_timeout! should use the default timeout if the timeout hasn't been set on the message instance" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- message.expects(:now).returns(0)
- message.set_timeout!
- assert_equal "#{Message::DEFAULT_HANDLER_TIMEOUT}", @store.get(message.msg_id, :timeout)
- message.expects(:now).returns(message.timeout)
- assert !message.timed_out?
- message.expects(:now).returns(Message::DEFAULT_HANDLER_TIMEOUT+1)
- assert message.timed_out?
- end
-
- test "incrementing execution attempts should increment by 1" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- assert_equal 1, message.increment_execution_attempts!
- assert_equal 2, message.increment_execution_attempts!
- assert_equal 3, message.increment_execution_attempts!
- end
-
- test "accessing execution attempts should return the number of execution attempts made so far" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- assert_equal 0, message.attempts
- message.increment_execution_attempts!
- assert_equal 1, message.attempts
- message.increment_execution_attempts!
- assert_equal 2, message.attempts
- message.increment_execution_attempts!
- assert_equal 3, message.attempts
- end
-
- test "accessing execution attempts should return 0 if none were made" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- assert_equal 0, message.attempts
- end
-
-
- test "attempts limit should be set exception limit + 1 iff the configured attempts limit is equal to or smaller than the exceptions limit" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :exceptions => 1, :store => @store)
- assert_equal 2, message.attempts_limit
- assert_equal 1, message.exceptions_limit
- message = Message.new("somequeue", header, 'foo', :exceptions => 2, :store => @store)
- assert_equal 3, message.attempts_limit
- assert_equal 2, message.exceptions_limit
- message = Message.new("somequeue", header, 'foo', :attempts => 5, :exceptions => 2, :store => @store)
- assert_equal 5, message.attempts_limit
- assert_equal 2, message.exceptions_limit
- end
-
- test "attempts limit should be reached after incrementing the attempt limit counter 'attempts limit' times" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :attempts =>2, :store => @store)
- assert !message.attempts_limit_reached?
- message.increment_execution_attempts!
- assert !message.attempts_limit_reached?
- message.increment_execution_attempts!
- assert message.attempts_limit_reached?
- message.increment_execution_attempts!
- assert message.attempts_limit_reached?
- end
-
- test "incrementing exception counts should increment by 1" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- assert_equal 1, message.increment_exception_count!
- assert_equal 2, message.increment_exception_count!
- assert_equal 3, message.increment_exception_count!
- end
-
- test "default exceptions limit should be reached after incrementing the attempt limit counter 1 time" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- assert !message.exceptions_limit_reached?
- message.increment_exception_count!
- assert message.exceptions_limit_reached?
- end
-
- test "exceptions limit should be reached after incrementing the attempt limit counter 'exceptions limit + 1' times" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :exceptions => 1, :store => @store)
- assert !message.exceptions_limit_reached?
- message.increment_exception_count!
- assert !message.exceptions_limit_reached?
- message.increment_exception_count!
- assert message.exceptions_limit_reached?
- message.increment_exception_count!
- assert message.exceptions_limit_reached?
- end
-
- test "failure to aquire a mutex should delete it from the database" do
- header = header_with_params({})
- message = Message.new("somequeue", header, 'foo', :store => @store)
- assert message.aquire_mutex!
- assert !message.aquire_mutex!
- assert !@store.exists(message.msg_id, :mutex)
- end
- end
-
-
class RoutingKeyTest < Minitest::Test
test "returns the routing key" do
header = header_with_params({})

ms precision for settings and exponential back off

- transformed time based values in settings to float to gain ms
  resolution
- added setting for fuzzied exponential back off