Add sharding (#11)

.github/workflows/test.yml

@@ -5,7 +5,7 @@ jobs:
  runs-on: ubuntu-latest
  strategy:
  matrix:
- ruby: ['2.6', '2.7', '3.0']
+ ruby: ['2.7', '3.0', '3.2']
  redis:
  - redis:5.0
  - redis:6.0
@@ -26,7 +26,6 @@ jobs:
  path: vendor/bundler
  key: ${{ hashFiles('Gemfile.lock') }}-${{ matrix.ruby }}
  - run: |
- gem install bundler
  bundle install --path=vendor/bundler
  bundle exec rspec
  bundle exec rubocop

CHANGELOG.md

@@ -1,5 +1,8 @@
 # CHANGELOG
 
+## v0.6.0
+* Allow sharding records into multiple streams
+
 ## v0.5.0
 * No longer delete delay messages
 

Gemfile

@@ -7,7 +7,6 @@ gem "bundler"
 gem "concurrent-ruby"
 gem "database_cleaner"
 gem "factory_bot"
-gem "mocha"
 gem "rake"
 gem "rspec"
 gem "rspec-instafail"

README.md

@@ -231,6 +231,33 @@ array of records to `Redstream::Producer#bulk`, like shown above. If you pass
 an `ActiveRecord::Relation`, the `#bulk` method will convert it to an array,
 i.e. load the whole result set into memory.
 
+## Sharding
+
+When you want to attach multiple consumers to a single stream, you maybe want
+to add sharding. This can be accomplished by specifying a dynamic stream name
+where you compute the shard key by hashing the primary key.
+
+```ruby
+class Product < ActiveRecord::Base
+ include Redstream::Model
+
+ NUM_SHARDS = 4
+
+ def self.redstream_name(shard)
+ "products-#{shard}"
+ end
+
+ def redstream_name
+ self.class.redstream_name(Digest::SHA1.hexdigest(id.to_s)[0, 4].to_i(16) % NUM_SHARDS)
+ end
+end
+```
+
+The sharding via hashing the primary key is neccessary, because we want each
+change of a specific object to end up in the same stream. Otherwise the order
+of changes for a specific object gets mixed up. Subsequently, you can add
+consumers, etc for each individual stream name.
+
 ## Namespacing
 
 In case you are using a shared redis, where multiple appications read/write

lib/redstream/consumer.rb

@@ -90,7 +90,7 @@ def run_once(&block)
  commit offset
  end
 
- sleep(5) unless got_lock
+ @lock.wait(5) unless got_lock
  rescue StandardError => e
  @logger.error e
 

lib/redstream/lock.rb

@@ -21,40 +21,67 @@ module Redstream
  # end
 
  class Lock
+ class Signal
+ def initialize
+ @mutex = Mutex.new
+ @condition_variable = ConditionVariable.new
+ end
+
+ def wait(timeout)
+ @mutex.synchronize { @condition_variable.wait(@mutex, timeout) }
+ end
+
+ def signal
+ @condition_variable.signal
+ end
+ end
+
  def initialize(name:)
  @name = name
  @id = SecureRandom.hex
  end
 
  def acquire(&block)
  got_lock = get_lock
- keep_lock(&block) if got_lock
+
+ if got_lock
+ keep_lock(&block)
+ release_lock
+ end
+
  got_lock
  end
 
+ def wait(timeout)
+ @wait_redis ||= Redstream.connection_pool.with(&:dup)
+ @wait_redis.brpop("#{Redstream.lock_key_name(@name)}.notify", timeout: timeout)
+ end
+
  private
 
  def keep_lock(&block)
- stop = false
- mutex = Mutex.new
+ stopped = false
+ signal = Signal.new
 
- Thread.new do
- until mutex.synchronize { stop }
- Redstream.connection_pool.with { |redis| redis.expire(Redstream.lock_key_name(@name), 5) }
+ thread = Thread.new do
+ until stopped
+ Redstream.connection_pool.with do |redis|
+ redis.expire(Redstream.lock_key_name(@name), 5)
+ end
 
- sleep 3
+ signal.wait(3)
  end
  end
 
  block.call
  ensure
- mutex.synchronize do
-  stop = true
- end
+ stopped = true
+ signal&.signal
+ thread&.join
  end
 
  def get_lock
- @get_lock_script = <<~GET_LOCK_SCRIPT
+ @get_lock_script = <<~SCRIPT
  local lock_key_name, id = ARGV[1], ARGV[2]
 
  local cur = redis.call('get', lock_key_name)
@@ -70,9 +97,30 @@ def get_lock
  end
 
  return false
- GET_LOCK_SCRIPT
+ SCRIPT
+
+ Redstream.connection_pool.with do |redis|
+ redis.eval(@get_lock_script, argv: [Redstream.lock_key_name(@name), @id])
+ end
+ end
+
+ def release_lock
+ @release_lock_script = <<~SCRIPT
+ local lock_key_name, id = ARGV[1], ARGV[2]
 
- Redstream.connection_pool.with { |redis| redis.eval(@get_lock_script, argv: [Redstream.lock_key_name(@name), @id]) }
+ local cur = redis.call('get', lock_key_name)
+
+ if cur and cur == id then
+ redis.call('del', lock_key_name)
+ end
+
+ redis.call('del', lock_key_name .. '.notify')
+ redis.call('rpush', lock_key_name .. '.notify', '1')
+ SCRIPT
+
+ Redstream.connection_pool.with do |redis|
+ redis.eval(@release_lock_script, argv: [Redstream.lock_key_name(@name), @id])
+ end
  end
  end
 end

lib/redstream/model.rb

@@ -58,5 +58,28 @@ def redstream_name
  def redstream_payload
  { id: id }
  end
+
+ # Override to customize the stream name. By default, the stream name
+ # is determined by the class name. If you override the instance method,
+ # please also override the class method.
+ #
+ # @example Sharding
+ # class Product
+ # include Redstream::Model
+ #
+ # NUM_SHARDS = 4
+ #
+ # def redstream_name
+ # self.class.redstream_name(Digest::SHA1.hexdigest(id.to_s)[0, 4].to_i(16) % NUM_SHARDS)
+ # end
+ #
+ # def self.redstream_name(shard)
+ # "products-#{shard}
+ # end
+ # end
+
+ def redstream_name
+ self.class.redstream_name
+ end
  end
 end

lib/redstream/producer.rb

@@ -31,7 +31,6 @@ class Producer
 
  def initialize(wait: false)
  @wait = wait
- @stream_name_cache = {}
 
  super()
  end
@@ -72,7 +71,7 @@ def bulk_delay(records)
  Redstream.connection_pool.with do |redis|
  redis.pipelined do |pipeline|
  slice.each do |object|
- pipeline.xadd(Redstream.stream_key_name("#{stream_name(object)}.delay"), { payload: JSON.dump(object.redstream_payload) })
+ pipeline.xadd(Redstream.stream_key_name("#{object.redstream_name}.delay"), { payload: JSON.dump(object.redstream_payload) })
  end
  end
  end
@@ -96,7 +95,7 @@ def bulk_queue(records)
  Redstream.connection_pool.with do |redis|
  redis.pipelined do |pipeline|
  slice.each do |object|
- pipeline.xadd(Redstream.stream_key_name(stream_name(object)), { payload: JSON.dump(object.redstream_payload) })
+ pipeline.xadd(Redstream.stream_key_name(object.redstream_name), { payload: JSON.dump(object.redstream_payload) })
  end
  end
  end
@@ -115,7 +114,7 @@ def bulk_queue(records)
 
  def delay(object)
  Redstream.connection_pool.with do |redis|
- res = redis.xadd(Redstream.stream_key_name("#{stream_name(object)}.delay"), { payload: JSON.dump(object.redstream_payload) })
+ res = redis.xadd(Redstream.stream_key_name("#{object.redstream_name}.delay"), { payload: JSON.dump(object.redstream_payload) })
  redis.wait(@wait, 0) if @wait
  res
  end
@@ -129,18 +128,10 @@ def delay(object)
 
  def queue(object)
  Redstream.connection_pool.with do |redis|
- redis.xadd(Redstream.stream_key_name(stream_name(object)), { payload: JSON.dump(object.redstream_payload) })
+ redis.xadd(Redstream.stream_key_name(object.redstream_name), { payload: JSON.dump(object.redstream_payload) })
  end
 
  true
  end
-
- private
-
- def stream_name(object)
- synchronize do
- @stream_name_cache[object.class] ||= object.class.redstream_name
- end
- end
  end
 end

lib/redstream/version.rb

@@ -1,3 +1,3 @@
 module Redstream
- VERSION = "0.5.0"
+ VERSION = "0.6.0"
 end

spec/redstream/consumer_spec.rb

@@ -86,5 +86,44 @@
 
  expect(redis.get(Redstream.offset_key_name(stream_name: "products", consumer_name: "consumer"))).to eq(all_messages.last[0])
  end
+
+ it "does not starve" do
+ product = create(:product)
+ stopped = false
+ results = Concurrent::Array.new
+
+ thread1 = Thread.new do
+ until stopped
+ Redstream::Consumer.new(name: "consumer", stream_name: "products").run_once do
+ results.push("thread1")
+
+ product.touch
+
+ sleep 0.1
+ end
+ end
+ end
+
+ thread2 = Thread.new do
+ until stopped
+ Redstream::Consumer.new(name: "consumer", stream_name: "products").run_once do
+ results.push("thread2")
+
+ product.touch
+
+ sleep 0.1
+ end
+ end
+ end
+
+ sleep 6
+
+ stopped = true
+ [thread1, thread2].each(&:join)
+
+ expect(results.size).to be > 10
+ expect(results.count("thread1").to_f / results.size).to be > 0.2
+ expect(results.count("thread2").to_f / results.size).to be > 0.2
+ end
  end
 end