Make delayed status key expiration optional
MODIFY
RELEASE_NOTES.rdoc
RELEASE_NOTES.rdoc
the dedup store for a configurable time window. The config option
is named redis_status_key_expiry_interval. Note that this will
significantly increase the cpu load and memory usage of the Redis
- server for applications using a large number of non-redundant messages.
+ server for applications using a large number of non-redundant
+ messages. This feature is turned off by default, but we will
+ activate it with the next major release (4.0.0).
== Version 3.2.0
* added currently processed message to the handler pre-processing step

MODIFY
lib/beetle/configuration.rb
lib/beetle/configuration.rb
@@ -37,13 +37,14 @@ module Beetle
# handler timeout.
attr_accessor :redis_failover_timeout
- # how long we want status keys to survive after we have seen the second message of a
- # redundant message pair. Defaults to 5 minutes. Setting this to a high value (hours)
- # will reduce the likelihood of executing handler logic more than once, but can cause
- # a higher redis database size with all associated problems. A handler can be
- # executed more than once if the ack to the RabbitMQ server gets lost between the consumer
- # and the RabbitMQ instance. This happens extremely seldom in our environment, but we cannot
- # rule it out and we also don't quite understand (yet) why it happens.
+ # how long we want status keys to survive after we have seen the second message of a redundant
+ # message pair. Defaults to 0 seconds, but will be set to something non-zero with the next
+ # major beetle release. A recommended value would be 5 minutes (300 seconds). Setting this to a
+ # high value (hours) will reduce the likelihood of executing handler logic more than once, but
+ # can cause a higher redis database size with all associated problems. A handler can be
+ # executed more than once if the ack to the RabbitMQ server gets lost between the consumer and
+ # the RabbitMQ instance. This happens extremely seldom in our environment, but we cannot rule
+ # it out and we also don't quite understand (yet) why it happens.
attr_accessor :redis_status_key_expiry_interval
# how often heartbeat messages are exchanged between failover
@@ -151,7 +152,7 @@ module Beetle
self.redis_servers = ""
self.redis_db = 4
self.redis_failover_timeout = 180.seconds
- self.redis_status_key_expiry_interval = 5.minutes
+ self.redis_status_key_expiry_interval = 0.seconds
self.redis_failover_client_heartbeat_interval = 10.seconds
self.redis_failover_client_dead_interval = 60.seconds

MODIFY
lib/beetle/deduplication_store.rb
lib/beetle/deduplication_store.rb
@@ -62,12 +62,14 @@ module Beetle
# store completion status for given <tt>msg_id</tt> if it doesn't exist yet. Returns whether the
# operation was successful.
def setnx_completed!(msg_id)
+ expiry = @config.redis_status_key_expiry_interval
+ return true if expiry == 0
with_failover do
redis.set(
key(msg_id, :status),
"completed",
:nx => true,
- :ex => @config.redis_status_key_expiry_interval,
+ :ex => expiry,
)
end
end
@@ -107,11 +109,12 @@ module Beetle
# delete all keys associated with the given <tt>msg_id</tt>.
def del_keys(msg_id)
+ expiry = @config.redis_status_key_expiry_interval
keys = keys(msg_id)
- status_key = keys.shift
+ status_key = keys.shift if expiry > 0
with_failover do
redis.del(*keys)
- redis.expire(status_key, @config.redis_status_key_expiry_interval)
+ redis.expire(status_key, expiry) if expiry > 0
end
end

MODIFY
test/beetle/message_test.rb
test/beetle/message_test.rb
@@ -129,7 +129,24 @@ module Beetle
end
end
+ test "successful processing of a non redundant message should delete all keys from the database" do
+ header = header_with_params({})
+ header.expects(:ack)
+ message = Message.new("somequeue", header, 'foo', :store => @store)
+ message.stubs(:simple?).returns(false)
+
+ assert !message.expired?
+ assert !message.redundant?
+
+ message.process(@null_handler)
+ keys = @store.keys(message.msg_id)
+ keys.each do |key|
+ assert !@store.redis.exists(key)
+ end
+ end
+
test "successful processing of a non redundant message should delete all keys from the database (except the status key, which should be set to expire)" do
+ @config.redis_status_key_expiry_interval = 1.minutes
header = header_with_params({})
header.expects(:ack)
message = Message.new("somequeue", header, 'foo', :store => @store)
@@ -148,7 +165,26 @@ module Beetle
end
end
+ test "successful processing of a redundant message twice should delete all keys from the database" do
+ header = header_with_params({:redundant => true})
+ header.expects(:ack).twice
+ message = Message.new("somequeue", header, 'foo', :store => @store)
+
+ assert !message.expired?
+ assert message.redundant?
+ assert !message.simple?
+
+ message.process(@null_handler)
+ message.process(@null_handler)
+
+ keys = @store.keys(message.msg_id)
+ keys.each do |key|
+ assert !@store.redis.exists(key)
+ end
+ end
+
test "successful processing of a redundant message twice should delete all keys from the database (except the status key, which should be set to expire)" do
+ @config.redis_status_key_expiry_interval = 1.minutes
header = header_with_params({:redundant => true})
header.expects(:ack).twice
message = Message.new("somequeue", header, 'foo', :store => @store)
@@ -302,7 +338,8 @@ module Beetle
class SimpleMessageTest < Minitest::Test
def setup
- @store = DeduplicationStore.new
+ @config = Configuration.new
+ @store = DeduplicationStore.new(@config)
@store.flushdb
end
@@ -333,6 +370,26 @@ module Beetle
assert_equal RC::AttemptsLimitReached, message.process(handler)
end
+ test "when processing a simple message, the handler should be executed only once if status keys are used" do
+ @config.redis_status_key_expiry_interval = 1.minute
+ header = header_with_params({})
+ message = Message.new("somequeue", header, 'foo', :attempts => 1, :store => @store)
+
+ handler = mock("handler")
+ s = sequence("s")
+ handler.expects(:pre_process).with(message).in_sequence(s)
+ header.expects(:ack).in_sequence(s)
+ handler.expects(:call).in_sequence(s)
+ assert_equal RC::OK, message.process(handler)
+
+ handler2 = mock("handler")
+ s2 = sequence("s2")
+ handler2.expects(:pre_process).with(message).in_sequence(s2)
+ header.expects(:ack).in_sequence(s2)
+ handler2.expects(:call).in_sequence(s2).never
+ assert_equal RC::OK, message.process(handler2)
+ end
+
end
class HandlerCrashTest < Minitest::Test

Make delayed status key expiration optional